Fix giant lists

Former-commit-id: 93d6ac101e02ea562c87853549080eb1ac85a3c6

.gitignore

@@ -3,3 +3,5 @@ exec-rssmon3
 **.sw*
 **/testdata
 **/._*
+**/exec-*
+exec-*

handlers/podcast/main.go

@@ -0,0 +1,234 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"local/args"
+	"local/storage"
+	"log"
+	"net/http"
+	"os"
+	"path"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/mmcdole/gofeed"
+)
+
+const sessionHeader = "X-Transmission-Session-Id"
+
+type Config struct {
+	url      string
+	vpntor   string
+	outdir   string
+	interval time.Duration
+	last     time.Time
+	db       storage.DB
+	ctx      context.Context
+	can      context.CancelFunc
+}
+
+func main() {
+	config, err := config()
+	if err != nil {
+		panic(err)
+	}
+	log.Println(config)
+
+	for {
+		if err := mainLoop(config); err != nil {
+			panic(err)
+		}
+	}
+}
+
+func mainLoop(config *Config) error {
+	block := config.interval - time.Since(config.last)
+	log.Printf("Blocking %v", block)
+	select {
+	case <-time.After(block):
+		if err := pull(config.db, config.vpntor, config.outdir, config.url); err != nil {
+			log.Println(err)
+		}
+		config.last = time.Now()
+	case <-config.ctx.Done():
+		if err := config.ctx.Err(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func config() (*Config, error) {
+	as := args.NewArgSet()
+	as.Append(args.STRING, "url", "url of rss feed", "http://192.168.0.86:33419/api/tag/podcast")
+	as.Append(args.STRING, "vpntor", "url of vpntor", "http://192.168.0.86:9091/transmission/rpc")
+	as.Append(args.DURATION, "interval", "interval to check feed", "30m")
+	as.Append(args.STRING, "outdir", "save dir", "/data/completed-rss")
+	as.Append(args.STRING, "db", "db type", "map")
+	as.Append(args.STRING, "addr", "db addr", "")
+	as.Append(args.STRING, "user", "db user", "")
+	as.Append(args.STRING, "pass", "db pass", "")
+	if err := as.Parse(); err != nil {
+		return &Config{}, err
+	}
+
+	db, err := storage.New(
+		storage.TypeFromString(as.Get("db").GetString()),
+		as.Get("addr").GetString(),
+		as.Get("user").GetString(),
+		as.Get("pass").GetString(),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	ctx, can := context.WithCancel(context.Background())
+	return &Config{
+		url:      as.Get("url").GetString(),
+		vpntor:   as.Get("vpntor").GetString(),
+		interval: as.Get("interval").GetDuration(),
+		outdir:   as.Get("outdir").GetString(),
+		db:       db,
+		ctx:      ctx,
+		can:      can,
+	}, nil
+}
+
+func pull(db storage.DB, vpntor, outdir, url string) error {
+	gofeed, err := getGoFeed(url)
+	if err != nil {
+		return err
+	}
+	log.Printf("feed: %v", gofeed.Title)
+	for _, item := range gofeed.Items {
+		if ok, err := isDone(db, item.Link); err != nil {
+			return err
+		} else if ok {
+			continue
+		}
+		s, err := getItemContent(item)
+		if err != nil {
+			return err
+		}
+		if err := handle(vpntor, outdir, s); err != nil {
+			return err
+		}
+		if err := db.Set(item.Link, []byte{}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func getGoFeed(url string) (*gofeed.Feed, error) {
+	resp, err := http.Get(url)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	return gofeed.NewParser().Parse(resp.Body)
+}
+
+func getItemContent(item *gofeed.Item) (string, error) {
+	s := item.Description
+	if s == "" {
+		s = item.Content
+	}
+	if s == "" {
+		resp, err := http.Get(item.Link)
+		if err != nil {
+			return s, err
+		}
+		defer resp.Body.Close()
+		b, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return s, err
+		}
+		s = string(b)
+	}
+	return s, nil
+}
+
+func isDone(db storage.DB, url string) (bool, error) {
+	_, err := db.Get(url)
+	if err == storage.ErrNotFound {
+		return false, nil
+	}
+	return true, err
+}
+
+func handle(vpntor, outdir, content string) error {
+	links := findMagnets(content)
+	sort.Strings(links)
+	for i := range links {
+		link := links[i]
+		if i > 0 && link == links[i-1] {
+			continue
+		}
+		log.Println(link)
+		if err := fetch(link, outdir); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func findMagnets(s string) []string {
+	magnetRegexp := regexp.MustCompile(`http[^"]*[0-9]+\.mp3`)
+	return magnetRegexp.FindAllString(s, -1)
+}
+
+func fetch(link, outdir string) error {
+	out := path.Join(outdir, path.Base(link))
+	if _, err := os.Stat(out); err == nil {
+		return nil
+	} else if os.IsNotExist(err) {
+	} else if err != nil {
+		return err
+	}
+
+	resp, err := http.Get(link)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	f, err := os.Create(out)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	io.Copy(f, resp.Body)
+	return nil
+}
+
+func buildReqBody(outdir, magnet string) io.Reader {
+	return strings.NewReader(fmt.Sprintf(`
+      {
+         "method": "torrent-add",
+         "arguments": {
+            "filename": %q,
+            "download-dir": %q
+         }
+      }
+   `, magnet, outdir))
+}
+
+func getSessionID(vpntor string) (string, error) {
+	resp, err := http.Get(vpntor)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	id := resp.Header.Get(sessionHeader)
+	if id == "" {
+		err = errors.New("session id header not found")
+	}
+	return id, err
+}

handlers/torrent/main.go

@@ -11,6 +11,7 @@ import (
 	"local/storage"
 	"log"
 	"net/http"
+	"path"
 	"regexp"
 	"strings"
 	"time"
@@ -99,12 +100,12 @@ func config() (*Config, error) {
 }
 
 func pull(db storage.DB, vpntor, outdir, url string) error {
-	gofeed, err := getGoFeed(url)
+	gfeed, err := getGoFeed(url)
 	if err != nil {
 		return err
 	}
-	log.Printf("feed: %v", gofeed.Title)
+	log.Printf("feed: %v", gfeed.Title)
-	for _, item := range gofeed.Items {
+	for _, item := range gfeed.Items {
 		if ok, err := isDone(db, item.Link); err != nil {
 			return err
 		} else if ok {
@@ -114,7 +115,10 @@ func pull(db storage.DB, vpntor, outdir, url string) error {
 		if err != nil {
 			return err
 		}
-		if err := handle(vpntor, outdir, s); err != nil {
+		if item.Author == nil {
+			item.Author = &gofeed.Person{Name: "."}
+		}
+		if err := handle(vpntor, path.Join(outdir, item.Author.Name), s); err != nil {
 			return err
 		}
 		if err := db.Set(item.Link, []byte{}); err != nil {

rss/feed.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"regexp"
 	"sort"
+	"strconv"
 	"time"
 
 	"github.com/mmcdole/gofeed"
@@ -21,6 +22,7 @@ type Feed struct {
 	TitleFilter   string
 	ContentFilter string
 	Tags          []string
+	Copyright     string
 }
 
 func SubmitFeed(f *Feed) error {
@@ -72,7 +74,7 @@ func (f *Feed) Pull() error {
 
 	itemTSs := []*time.Time{}
 	for _, i := range gofeed.Items {
-		item, err := newItem(i, f.ContentFilter)
+		item, err := newItem(i, f.ContentFilter, f.Copyright)
 		if err != nil {
 			log.Println("[Pull]", err)
 			continue
@@ -83,9 +85,10 @@ func (f *Feed) Pull() error {
 			continue
 		}
 		if ok := regexp.MustCompile(f.TitleFilter).MatchString(item.Title); !ok {
-			log.Println("[Pull]", "Skipping bad titled item")
+			log.Printf("[Pull] Skipping bad titled item: %v doesn't match /%v/", item.Title, f.TitleFilter)
 			continue
 		}
+		log.Printf("[Pull] Saving item %v for %v /%v/", f.Key, f.URL, f.TitleFilter)
 		if err := item.save(f.Key); err != nil {
 			log.Println("[Pull]", err)
 			continue
@@ -123,7 +126,7 @@ func (f *Feed) Items(limit int) ([]*Item, error) {
 }
 
 func (f *Feed) List(limit int) ([]string, error) {
-	keys, err := config.Values().DB.List([]string{nsItems, f.Key})
+	keys, err := config.Values().DB.List([]string{nsItems, f.Key}, " ", "}}}}}", strconv.Itoa(limit), "-")
 	if err != nil {
 		return nil, err
 	}

rss/feed_test.go

@@ -177,3 +177,37 @@ func TestRSSFeedPull(t *testing.T) {
 		t.Fatal(i)
 	}
 }
+
+func TestRSSFeedListLimitedDescending(t *testing.T) {
+	initRSSFeed()
+
+	s := mockRSS()
+	defer s.Close()
+
+	f := newFeed("key")
+	f.TitleFilter = "50."
+	f.ContentFilter = "b"
+	f.Tags = []string{"c"}
+	f.URL = s.URL
+
+	log.SetOutput(bytes.NewBuffer(nil))
+	defer log.SetOutput(os.Stderr)
+	if err := f.Pull(); err != nil {
+		t.Fatal(err)
+	}
+	log.SetOutput(os.Stderr)
+
+	itemKeys, err := f.List(5)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(itemKeys) != 5 {
+		t.Fatal(len(itemKeys))
+	}
+
+	for i := range itemKeys {
+		if i > 0 && itemKeys[i] > itemKeys[i-1] {
+			t.Error(itemKeys[i], ">", itemKeys[i-1])
+		}
+	}
+}

rss/item.go

@@ -20,16 +20,18 @@ type Item struct {
 	Link      string
 	Content   string
 	TS        time.Time
+	Copyright string
 }
 
 type Items []*Item
 
-func newItem(i *gofeed.Item, contentFilter string) (*Item, error) {
+func newItem(i *gofeed.Item, contentFilter, copyright string) (*Item, error) {
 	item := &Item{
 		Title:     i.Title,
 		Link:      i.Link,
 		Content:   i.Content,
 		TS:        latestTSPtr(i.UpdatedParsed, i.PublishedParsed),
+		Copyright: copyright,
 	}
 
 	if item.Content == "" {
@@ -47,6 +49,9 @@ func newItem(i *gofeed.Item, contentFilter string) (*Item, error) {
 		}
 		item.Content = string(b)
 	}
+	for _, enclosure := range i.Enclosures {
+		item.Content += fmt.Sprintf(`<br><a href="%s">%s</a>`, enclosure.URL, enclosure.URL)
+	}
 	if unescaped, err := url.QueryUnescape(item.Content); err == nil {
 		item.Content = unescaped
 	}

rss/item_test.go

@@ -30,12 +30,12 @@ func TestRSSItemNewEncodeDecode(t *testing.T) {
 	gofeed.Items[0].Content = ""
 	gofeed.Items[0].Description = ""
 
-	item, err := newItem(gofeed.Items[0], ".*")
+	item, err := newItem(gofeed.Items[0], ".*", "")
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	itemB, err := newItem(gofeed.Items[0], "Podcast")
+	itemB, err := newItem(gofeed.Items[0], "Podcast", "")
 	if err != nil {
 		t.Fatal(err)
 	}

rss/serialize.go

@@ -43,6 +43,7 @@ func WriteFeed(w io.Writer, tag string, items []*Item) error {
 			Title:       item.Title,
 			Link:        &feeds.Link{Href: item.Link},
 			Description: item.Content,
+			Author:      &feeds.Author{Name: item.Copyright},
 			Created:     item.TS,
 		}
 	}

rss/serialize_test.go

@@ -47,6 +47,7 @@ func TestRSSWriteFeed(t *testing.T) {
 	f.ContentFilter = "b"
 	f.Tags = []string{"c"}
 	f.URL = s.URL
+	f.Copyright = "copyright"
 
 	log.SetOutput(bytes.NewBuffer(nil))
 	defer log.SetOutput(os.Stderr)

rss/testdata/rss.xml.REMOVED.git-id

@@ -1 +0,0 @@
-d36f6c94dfbbaaeac339bfaec9d0dd13b0ff099b

server/routes.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"net/http"
 	"regexp"
+	"strconv"
 	"time"
 )
 
@@ -50,6 +51,10 @@ func (s *Server) tag(w http.ResponseWriter, r *http.Request) {
 		s.notFound(w, r)
 		return
 	}
+	limit, err := strconv.Atoi(r.URL.Query().Get("n"))
+	if err != nil {
+		limit = 20
+	}
 	tag := regexp.MustCompile("^.*\\/").ReplaceAllString(r.URL.Path, "")
 	feeds, err := rss.TaggedFeeds(tag)
 	if err != nil {
@@ -58,7 +63,7 @@ func (s *Server) tag(w http.ResponseWriter, r *http.Request) {
 	}
 	items := []*rss.Item{}
 	for _, feed := range feeds {
-		feedItems, err := feed.Items(20)
+		feedItems, err := feed.Items(limit)
 		if err != nil {
 			s.error(w, r, err)
 		}
@@ -79,6 +84,7 @@ func (s *Server) feed(w http.ResponseWriter, r *http.Request) {
 		URL           string   `json:"url"`
 		Interval      string   `json:"refresh"`
 		TitleFilter   string   `json:"items"`
+		Copyright     string   `json:"copyright"`
 		ContentFilter string   `json:"content"`
 		Tags          []string `json:"tags"`
 	}
@@ -97,6 +103,7 @@ func (s *Server) feed(w http.ResponseWriter, r *http.Request) {
 		TitleFilter:   putFeed.TitleFilter,
 		ContentFilter: putFeed.ContentFilter,
 		Tags:          putFeed.Tags,
+		Copyright:     putFeed.Copyright,
 	}
 	if err := rss.SubmitFeed(f); err != nil {
 		s.error(w, r, err)