all

exchange/exchange.go

@@ -5,12 +5,15 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"local/logger"
+	"io"
 	"local/rssmon2/monitor"
 	"local/rssmon2/rss"
 	"local/rssmon2/server"
 	"local/rssmon2/store"
+	"log"
 	"net/http"
+	"os"
+	"path"
 	"regexp"
 	"sort"
 	"strings"
@@ -25,6 +28,7 @@ type Exchange struct {
 	Srv      *server.Server
 	allFeeds map[string]*rss.Feed
 	toraddr  string
+	savepath string
 }
 
 func New(mon *monitor.Monitor, sclient store.Client, srv *server.Server, toraddr string) *Exchange {
@@ -34,6 +38,7 @@ func New(mon *monitor.Monitor, sclient store.Client, srv *server.Server, toraddr
 		Srv:      srv,
 		allFeeds: make(map[string]*rss.Feed),
 		toraddr:  toraddr,
+		savepath: "/mnt/podcasts",
 	}
 }
 
@@ -62,12 +67,12 @@ func (ex *Exchange) LoadDB() error {
 func (ex *Exchange) NewFeed(url, itemFilter, contentFilter string, tags []string, interval time.Duration) {
 	feed, err := rss.New(url, itemFilter, contentFilter, tags, interval)
 	if err != nil {
-		logger.Logf("can't create new RSS %q: %v", url, err)
+		log.Printf("can't create new RSS %q: %v", url, err)
 		return
 	}
 	ex.allFeeds[url] = feed
 	if err := ex.Mon.Submit(url, feed.Interval); err != nil {
-		logger.Logf("Cannot accept new feed %q: %v", url, err)
+		log.Printf("Cannot accept new feed %q: %v", url, err)
 	}
 }
 
@@ -106,6 +111,32 @@ func (ex *Exchange) GetFeedItem(ID string) (string, error) {
 	return item.Content, nil
 }
 
+func (ex *Exchange) ListTag(tag string, n int) (string, error) {
+	feeds := []*rss.Feed{}
+	feedNames, err := ex.SClient.List(nsForFeeds, "", true, -1)
+	if err != nil {
+		return "", err
+	}
+	for _, feedName := range feedNames {
+		b, err := ex.SClient.Get(nsForFeeds, feedName)
+		if err != nil {
+			return "", err
+		}
+		feed, err := rss.Deserialize(b)
+		if err != nil {
+			return "", err
+		}
+
+		for i := range feed.Tags {
+			if feed.Tags[i] == tag {
+				feeds = append(feeds, feed)
+			}
+		}
+	}
+	b, err := json.Marshal(feeds)
+	return string(b), err
+}
+
 func (ex *Exchange) GetFeedTagRSS(tag string) (string, error) {
 	feedNames, err := ex.SClient.List(nsForFeeds, "", true, -1)
 	if err != nil {
@@ -159,48 +190,48 @@ func (ex *Exchange) UpdateFeed(url string) {
 	if !ok {
 		f, err := rss.New(url, "", "", nil, time.Minute)
 		if err != nil {
-			logger.Logf("cannot identify unknown feed triggered in monitor: %q: %v", url, err)
+			log.Printf("cannot identify unknown feed triggered in monitor: %q: %v", url, err)
 			return
 		}
 		b, err := ex.SClient.Get(nsForFeeds, f.ID())
 		if err != nil {
-			logger.Logf("cannot get unknown feed triggered in monitor: %q: %v", url, err)
+			log.Printf("cannot get unknown feed triggered in monitor: %q: %v", url, err)
 			return
 		}
 		feed, err = rss.Deserialize(b)
 		if err != nil {
-			logger.Logf("cannot deserialize feed triggered in monitor: %q: %v", url, err)
+			log.Printf("cannot deserialize feed triggered in monitor: %q: %v", url, err)
 			return
 		}
 	}
 	items, err := ex.allFeeds[url].Update()
 	if err != nil {
-		logger.Logf("can't update old RSS %q: %v", url, err)
+		log.Printf("can't update old RSS %q: %v", url, err)
 		return
 	}
 	b, err := feed.Serialize()
 	if err != nil {
-		logger.Logf("can't serialize to save RSS %q: %v", url, err)
+		log.Printf("can't serialize to save RSS %q: %v", url, err)
 		return
 	}
 	if err := ex.SClient.Set(nsForFeeds, feed.ID(), b); err != nil {
-		logger.Logf("can't save RSS %q.%q: %v", nsForFeeds, feed.ID(), err)
+		log.Printf("can't save RSS %q.%q: %v", nsForFeeds, feed.ID(), err)
 		return
 	}
-	logger.Log("Saved feed", feed)
+	log.Print("Saved feed", feed)
 	for i := range items {
 		b, err := items[i].Serialize()
 		if err != nil {
-			logger.Logf("can't save rss item %q.%q: %v", url, items[i].Link, err)
+			log.Printf("can't save rss item %q.%q: %v", url, items[i].Link, err)
 			return
 		}
 		if err := ex.SClient.Set(feed.ID(), items[i].ID(), b); err != nil {
-			logger.Logf("can't save rss item %q.%q: %v", feed.ID(), items[i].ID(), err)
+			log.Printf("can't save rss item %q.%q: %v", feed.ID(), items[i].ID(), err)
 			return
 		}
-		//logger.Log("Saved feed item", feed.ID(), items[i].ID(), items[i])
+		//log.Print("Saved feed item", feed.ID(), items[i].ID(), items[i])
 	}
-	logger.Logf("Saved %d feed items for %s", len(items), feed.Title)
+	log.Printf("Saved %d feed items for %s", len(items), feed.Title)
 	go ex.handlerByTag(feed.Tags, items)
 }
 
@@ -210,21 +241,60 @@ func (ex *Exchange) handlerByTag(tags []string, items []*rss.Item) {
 			switch tags[i] {
 			case "torrent":
 				ex.handlerTorrent(items[j])
+			case "podcast":
+				ex.handlerPodcast(items[j])
 			}
 		}
 	}
 }
 
+func (ex *Exchange) handlerPodcast(item *rss.Item) {
+	os.MkdirAll(ex.savepath, os.ModePerm)
+
+	if len(item.Enclosures) < 1 {
+		return
+	}
+	for i := range item.Enclosures {
+		link := item.Enclosures[i]
+		link = strings.TrimSpace(link)
+		if !strings.Contains(link, ".mp3") {
+			continue
+		}
+		savePath := path.Join(ex.savepath, path.Base(link))
+		if _, err := os.Stat(savePath); !os.IsNotExist(err) {
+			log.Printf("err: %v already exists", savePath)
+			continue
+		}
+
+		resp, err := http.Get(link)
+		if err != nil {
+			log.Printf("cannot get podcast %q: %v", link, err)
+			continue
+		}
+		defer resp.Body.Close()
+
+		out, err := os.Create(savePath)
+		if err != nil {
+			log.Printf("cannot create file %q for saving: %v", savePath, err)
+			continue
+		}
+		if _, err := io.Copy(out, resp.Body); err != nil {
+			log.Printf("failed to write: %v", err)
+			continue
+		}
+	}
+}
+
 func (ex *Exchange) handlerTorrent(item *rss.Item) {
 	magnet := regexp.MustCompile("magnet:.xt[^ $]*")
 	if !magnet.MatchString(item.Content) {
-		logger.Log("no magnet link: ", item.Content)
+		log.Print("no magnet link: ", item.Content)
 	}
 	match := magnet.FindString(item.Content)
 
 	resp, err := http.Get(ex.toraddr)
 	if err != nil {
-		logger.Logf("cannot get session id at %v: %v", ex.toraddr, err)
+		log.Printf("cannot get session id at %v: %v", ex.toraddr, err)
 		return
 	}
 	sessionID := resp.Header.Get("X-Transmission-Session-Id")
@@ -232,31 +302,32 @@ func (ex *Exchange) handlerTorrent(item *rss.Item) {
 		"method": "torrent-add",
 		"arguments": map[string]interface{}{
 			"filename":     match,
+			"download-dir": "/data/completed-rss",
 		},
 	})
 	if err != nil {
-		logger.Log("cannot build request body: %v", err)
+		log.Print("cannot build request body: %v", err)
 		return
 	}
 	req, err := http.NewRequest("POST", ex.toraddr, bytes.NewBuffer(requestBody))
 	if err != nil {
-		logger.Log("cannot POST request: %v", err)
+		log.Print("cannot POST request: %v", err)
 		return
 	}
 	req.Header.Add("X-Transmission-Session-Id", sessionID)
 	resp, err = (&http.Client{}).Do(req)
 	if err != nil {
-		logger.Log("failed POST: %v", err)
+		log.Print("failed POST: %v", err)
 		return
 	}
 
 	var out map[string]interface{}
 	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
-		logger.Log("cannot decode response: %v", err)
+		log.Print("cannot decode response: %v", err)
 		return
 	}
 
 	if v, ok := out["result"]; !ok || v != "success" {
-		logger.Logf("failed magnet submission for feed item %v: %v", item.Name, v)
+		log.Printf("failed magnet submission for feed item %v: %v", item.Name, v)
 	}
 }

main.go

@@ -1,12 +1,12 @@
 package main
 
 import (
-	"local/logger"
 	"local/rssmon2/config"
 	"local/rssmon2/exchange"
 	"local/rssmon2/monitor"
 	"local/rssmon2/server"
 	"local/rssmon2/store"
+	"log"
 )
 
 func main() {
@@ -42,6 +42,7 @@ func core() {
 		ex.GetFeedRSS,
 		ex.GetFeedItem,
 		ex.GetFeedTagRSS,
+		ex.ListTag,
 	); err != nil {
 		panic(err)
 	}
@@ -51,7 +52,7 @@ func core() {
 		panic(err)
 	}
 
-	logger.Log("Starting with", config)
+	log.Print("Starting with", config)
 	if err := srv.Serve(); err != nil {
 		panic(err)
 	}

main_test.go

@@ -110,6 +110,11 @@ func Test_Core(t *testing.T) {
 			status: 200,
 			post:   func() { time.Sleep(time.Second * 15) },
 		},
+		{
+			method: "get",
+			path:   "api/list/tag/gotest",
+			status: 200,
+		},
 	}
 	for _, c := range cases {
 		c.method = strings.ToUpper(c.method)

monitor/monitor.go

@@ -2,7 +2,7 @@ package monitor
 
 import (
 	"errors"
-	"local/logger"
+	"log"
 	"time"
 
 	"github.com/golang-collections/go-datastructures/queue"
@@ -59,29 +59,29 @@ func (monitor *Monitor) loop() error {
 			}
 			var err error
 			if nextEvent, err = nextEventTime(queue); err != nil {
-				logger.Log("no next event time", err)
+				log.Print("no next event time", err)
 				nextEvent = time.Now().Add(time.Minute * 60)
 				continue
 			}
 		case <-time.After(nextEvent.Sub(time.Now())):
 			items, err := queue.Get(1)
 			if err != nil {
-				logger.Fatal("can't get item")
+				log.Fatal("can't get item")
 			}
 			if len(items) == 0 {
-				logger.Log("no items in queue")
+				log.Print("no items in queue")
 				nextEvent = time.Now().Add(time.Minute * 60)
 				continue
 			}
 			item, ok := items[0].(*Item)
 			if !ok {
-				logger.Fatal("queue contains illegal item")
+				log.Fatal("queue contains illegal item")
 			}
 			go monitor.trigger(item.URL)
 			item.increment()
 			queue.Put(item)
 			if nextEvent, err = nextEventTime(queue); err != nil {
-				logger.Log("no next event time", err)
+				log.Print("no next event time", err)
 				nextEvent = time.Now().Add(time.Minute * 60)
 				continue
 			}

rss/feed.go

@@ -5,7 +5,7 @@ import (
 	"encoding/gob"
 	"fmt"
 	"io/ioutil"
-	"local/logger"
+	"log"
 	"net/http"
 	"regexp"
 	"strings"
@@ -107,8 +107,11 @@ func (feed *Feed) fromGofeed(gofeed *gofeed.Feed) ([]*Item, error) {
 	if err != nil {
 		return nil, err
 	}
-	feed.Updated = latest.Add(time.Second) //time.Now() //*updated //time.Now().UTC() //*updated
+	if latest != feed.Updated {
+		feed.Updated = latest.Add(time.Second)
+	}
 	feed.Title = gofeed.Title
+	log.Printf("found %v new items for %v", len(newitems), feed.Title)
 	return newitems, nil
 }
 
@@ -118,19 +121,19 @@ func (feed *Feed) appendNewItems(items []*gofeed.Item) ([]*Item, time.Time, erro
 	for i := range items {
 		t := gofeedItemTS(items[i])
 		if *t != (time.Time{}) && t.Before(feed.Updated) {
-			logger.Logf("\tnot adding %v because its timestamp before %v", items[i].Link, feed.Updated)
+			log.Printf("\tnot adding %v because its timestamp (%v) before updated (%v)", items[i].Link, *t, feed.Updated)
 			continue
 		}
 		if t.After(latest) {
 			latest = *t
 		}
 		if ok, _ := regexp.MatchString(feed.ItemFilter, items[i].Title); !ok {
-			logger.Logf("\tnot adding %v because its title doesnt match item filter %v", items[i].Link, feed.ItemFilter)
+			//log.Printf("\tnot adding %v because its title doesnt match item filter %v", items[i].Link, feed.ItemFilter)
 			continue
 		}
 		item := FromGofeedItem(items[i], feed.ContentFilter)
 		newitems = append(newitems, item)
 	}
-	logger.Logf("\tnew latest is %v", latest)
+	log.Printf("\tnew latest is %v from %v", latest, feed.Updated)
 	return newitems, latest, nil
 }

rss/item.go

@@ -19,14 +19,16 @@ type Item struct {
 	Name       string
 	Link       string
 	Content    string
+	Enclosures []string
 	TS         time.Time
 }
 
 func (item *Item) String() string {
-	return fmt.Sprintf("Name %v, Link %v, Content %v, TS %v",
+	return fmt.Sprintf("Name %v, Link %v, Content %v, Enclosures %v, TS %v",
 		item.Name,
 		item.Link,
 		len(item.Content),
+		item.Enclosures,
 		item.TS.Local(),
 	)
 }
@@ -51,10 +53,15 @@ func (item *Item) Serialize() ([]byte, error) {
 }
 
 func FromGofeedItem(gfitem *gofeed.Item, filter string) *Item {
+	enclosures := []string{}
+	for i := range gfitem.Enclosures {
+		enclosures = append(enclosures, gfitem.Enclosures[i].URL)
+	}
 	item := &Item{
 		Name:       gfitem.Title,
 		Link:       gfitem.Link,
 		Content:    "",
+		Enclosures: enclosures,
 		TS:         *gofeedItemTS(gfitem),
 	}
 	content := gfitem.Content

rss/xml.go

@@ -20,7 +20,8 @@ func ToRSS(feed *Feed, items []*Item) (string, error) {
 		v.Add("url", feed.ID()+"."+items[i].ID())
 		root.Items[i] = &feeds.Item{
 			Title: items[i].Name,
-			Link:        &feeds.Link{Href: "/api/feed/item?" + v.Encode()},
+			//Link:        &feeds.Link{Href: "/api/feed/item?" + v.Encode()},
+			Link:        &feeds.Link{Href: items[i].Link},
 			Description: items[i].Content,
 			Created:     items[i].TS,
 		}

server/server.go

@@ -3,7 +3,7 @@ package server
 import (
 	"encoding/json"
 	"fmt"
-	"local/logger"
+	"log"
 	"net/http"
 	"net/url"
 	"os"
@@ -22,15 +22,24 @@ type Server struct {
 	getFeedHandler     func(string, int) (string, error)
 	getFeedItemHandler func(string) (string, error)
 	getFeedTagHandler  func(string) (string, error)
+	getListTagHandler  func(string, int) (string, error)
 }
 
-func New(addr string, newFeedHandler func(string, string, string, []string, time.Duration), getFeedHandler func(string, int) (string, error), getFeedItemHandler func(string) (string, error), getFeedTagHandler func(string) (string, error)) (*Server, error) {
+func New(
+	addr string,
+	newFeedHandler func(string, string, string, []string, time.Duration),
+	getFeedHandler func(string, int) (string, error),
+	getFeedItemHandler func(string) (string, error),
+	getFeedTagHandler func(string) (string, error),
+	getListTagHandler func(string, int) (string, error),
+) (*Server, error) {
 	return &Server{
 		addr:               addr,
 		newFeedHandler:     newFeedHandler,
 		getFeedHandler:     getFeedHandler,
 		getFeedItemHandler: getFeedItemHandler,
 		getFeedTagHandler:  getFeedTagHandler,
+		getListTagHandler:  getListTagHandler,
 	}, nil
 }
 
@@ -81,6 +90,8 @@ func (s *Server) api(w http.ResponseWriter, r *http.Request) {
 	switch advance(r) {
 	case "feed":
 		s.feed(w, r)
+	case "list":
+		s.list(w, r)
 	default:
 		s.notFound(w, r)
 	}
@@ -155,7 +166,7 @@ func (s *Server) getFeedTag(w http.ResponseWriter, r *http.Request) {
 	}
 	feedBody, err := s.getFeedTagHandler(url)
 	if err != nil {
-		logger.Logf("cannot get feed tag %s: %v", url, err)
+		log.Printf("cannot get feed tag %s: %v", url, err)
 		s.mybad(w, r)
 		return
 	}
@@ -170,7 +181,7 @@ func (s *Server) getFeedItem(w http.ResponseWriter, r *http.Request) {
 	}
 	feedBody, err := s.getFeedItemHandler(url)
 	if err != nil {
-		logger.Logf("cannot get feed item %s: %v", url, err)
+		log.Printf("cannot get feed item %s: %v", url, err)
 		s.mybad(w, r)
 		return
 	}
@@ -191,13 +202,31 @@ func (s *Server) getFeed(w http.ResponseWriter, r *http.Request) {
 	}
 	feedBody, err := s.getFeedHandler(url, limit)
 	if err != nil {
-		logger.Logf("cannot get feed %s: %v", url, err)
+		log.Printf("cannot get feed %s: %v", url, err)
 		s.mybad(w, r)
 		return
 	}
 	fmt.Fprintln(w, feedBody)
 }
 
+func (s *Server) list(w http.ResponseWriter, r *http.Request) {
+	switch advance(r) {
+	case "tag":
+		s.listTag(w, r)
+	default:
+		s.notFound(w, r)
+	}
+}
+
+func (s *Server) listTag(w http.ResponseWriter, r *http.Request) {
+	tag := advance(r)
+	out, err := s.getListTagHandler(tag, -1)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Fprintln(w, out)
+}
+
 func validURL(loc string) bool {
 	_, err := url.ParseRequestURI(loc)
 	return err == nil

server/server_test.go

@@ -19,7 +19,7 @@ func Test_Server(t *testing.T) {
 	testPort = strings.Split(server.Listener.Addr().String(), ":")[1]
 
 	var err error
-	s, err := New(testPort, func(string, string, string, []string, time.Duration) {}, func(string, int) (string, error) { return "", nil }, func(string) (string, error) { return "", nil }, func(string) (string, error) { return "", nil })
+	s, err := New(testPort, func(string, string, string, []string, time.Duration) {}, func(string, int) (string, error) { return "", nil }, func(string) (string, error) { return "", nil }, func(string) (string, error) { return "", nil }, func(string, int) (string, error) { return "", nil })
 	if err != nil {
 		t.Errorf("failed to create server: %v", err)
 	}
@@ -58,6 +58,12 @@ func Test_Server(t *testing.T) {
 	if err := checkStatus("GET", "api/feed/tag/b", http.StatusOK); err != nil {
 		t.Errorf(err.Error())
 	}
+	if err := checkStatus("GET", "api/feed/tag/b", http.StatusOK); err != nil {
+		t.Errorf(err.Error())
+	}
+	if err := checkStatus("GET", "api/list/tag/b", http.StatusOK); err != nil {
+		t.Errorf(err.Error())
+	}
 	syscall.Kill(syscall.Getpid(), syscall.SIGINT)
 }