Rssmon2/exchange/exchange.go

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) 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 {
		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)
	}
}