package exchange import ( "bytes" "encoding/json" "errors" "fmt" "io" "local/logger" "local/rssmon2/monitor" "local/rssmon2/rss" "local/rssmon2/server" "local/rssmon2/store" "log" "net/http" "os" "path" "regexp" "sort" "strings" "time" ) const nsForFeeds = "FEEDS" type Exchange struct { Mon *monitor.Monitor SClient store.Client 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 { return &Exchange{ Mon: mon, SClient: sclient, Srv: srv, allFeeds: make(map[string]*rss.Feed), toraddr: toraddr, savepath: "/mnt/podcasts", } } func (ex *Exchange) LoadDB() error { oldFeeds, err := ex.SClient.List(nsForFeeds, "", true, -1) if err != nil { return err } for _, feedID := range oldFeeds { b, err := ex.SClient.Get(nsForFeeds, feedID) if err != nil { return err } feed, err := rss.Deserialize(b) if err != nil { return err } ex.allFeeds[feed.Link] = feed if err := ex.Mon.Submit(feed.Link, feed.Interval); err != nil { return err } } return nil } 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) 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) } } func (ex *Exchange) GetFeedRSS(url string, n int) (string, error) { feed, ok := ex.allFeeds[url] if !ok { return "", errors.New("unknown feed " + url) } itemKeys, err := ex.SClient.List(feed.ID(), "", false, n) if err != nil { return "", err } items := make([]*rss.Item, len(itemKeys)) for i := range itemKeys { b, err := ex.SClient.Get(feed.ID(), itemKeys[i]) if err != nil { return "", errors.New("cannot get feed item " + itemKeys[i]) } items[i], err = rss.DeserializeItem(b) if err != nil { return "", errors.New("cannot deserialize feed item" + itemKeys[i]) } } return rss.ToRSS(feed, items) } func (ex *Exchange) GetFeedItem(ID string) (string, error) { b, err := ex.SClient.Get(strings.Split(ID, ".")[0], strings.Join(strings.Split(ID, ".")[1:], ".")) if err != nil { return "", errors.New("cannot get feed item " + ID) } item, err := rss.DeserializeItem(b) if err != nil { return "", errors.New("cannot deserialize feed item" + ID) } return item.Content, nil } func (ex *Exchange) GetFeedTagRSS(tag string) (string, error) { feedNames, err := ex.SClient.List(nsForFeeds, "", true, -1) if err != nil { return "", err } combinedItems := []*rss.Item{} 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 { itemKeys, err := ex.SClient.List(feed.ID(), "", false, 20) // TODO variable, or pick most recent n or something if err != nil { return "", err } for i := range itemKeys { b, err := ex.SClient.Get(feed.ID(), itemKeys[i]) if err != nil { return "", errors.New("cannot get feed item " + itemKeys[i]) } item, err := rss.DeserializeItem(b) if err != nil { return "", errors.New("cannot deserialize feed item" + itemKeys[i]) } combinedItems = append(combinedItems, item) } break } } } combinedFeed, err := rss.New(tag, "", "", nil, time.Minute) if err != nil { return "", err } sort.Slice(combinedItems, func(i, j int) bool { return !combinedItems[i].TS.Before(combinedItems[j].TS) }) combinedFeed.Title = fmt.Sprintf("T: %v", tag) return rss.ToRSS(combinedFeed, combinedItems) } func (ex *Exchange) UpdateFeed(url string) { feed, ok := ex.allFeeds[url] 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) 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) return } feed, err = rss.Deserialize(b) if err != nil { logger.Logf("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) return } b, err := feed.Serialize() if err != nil { logger.Logf("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) return } logger.Log("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) 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) return } //logger.Log("Saved feed item", feed.ID(), items[i].ID(), items[i]) } logger.Logf("Saved %d feed items for %s", len(items), feed.Title) go ex.handlerByTag(feed.Tags, items) } func (ex *Exchange) handlerByTag(tags []string, items []*rss.Item) { for i := range tags { for j := range items { 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) } 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) return } sessionID := resp.Header.Get("X-Transmission-Session-Id") requestBody, err := json.Marshal(map[string]interface{}{ "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) return } req, err := http.NewRequest("POST", ex.toraddr, bytes.NewBuffer(requestBody)) if err != nil { logger.Log("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) return } var out map[string]interface{} if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { logger.Log("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) } }