135 lines
3.3 KiB
Go
135 lines
3.3 KiB
Go
package app
|
|
|
|
import (
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
"go.balki.me/tss/log"
|
|
"go.balki.me/tss/parser"
|
|
"go.balki.me/tss/proxy"
|
|
"go.balki.me/tss/telegram"
|
|
)
|
|
|
|
func Run(configPath string) {
|
|
cfg, err := ParseConfig(configPath)
|
|
|
|
if err != nil {
|
|
log.Panic("failed to parse config", "path", configPath, "err", err)
|
|
}
|
|
|
|
scheduler, err := NewScheduler(cfg.LastSuccessPath)
|
|
if err != nil {
|
|
log.Panic("failed to create scheduler", "path", cfg.LastSuccessPath, "error", err)
|
|
}
|
|
|
|
defer func() {
|
|
if err := scheduler.Save(); err != nil {
|
|
log.Panic("failed to save last success info", "path", cfg.LastSuccessPath, "err", err)
|
|
}
|
|
}()
|
|
|
|
tgramProxy, err := proxy.GetTransport(cfg.TelegramProxy)
|
|
if err != nil {
|
|
log.Panic("failed to get proxy transport", "proxyURL", cfg.TelegramProxy, "error", err)
|
|
}
|
|
|
|
tgram := telegram.NewTelegramSender(tgramProxy, cfg.TelegramAuthToken)
|
|
|
|
for _, feed := range cfg.Feeds {
|
|
log.Info("processing feed", "feed", feed.Name)
|
|
ProcessFeed(feed, scheduler, cfg.DbDir, tgram)
|
|
}
|
|
}
|
|
|
|
func ProcessFeed(feed FeedCfg, scheduler Scheduler, dbDir string, tgram telegram.TelegramSender) {
|
|
sd, err := scheduler.ShouldDownload(feed.Name, feed.Cron)
|
|
if err != nil {
|
|
log.Error("shouldDownload failed", "feed", feed.Name, "err", err)
|
|
return
|
|
}
|
|
|
|
if !sd {
|
|
log.Info("skipping feed due to schedule", "feed", feed.Name)
|
|
return
|
|
}
|
|
|
|
db, err := NewDB(dbDir, feed.Name)
|
|
if err != nil {
|
|
log.Error("failed to get db", "feed", feed.Name, "db_dir", dbDir, "error", err)
|
|
return
|
|
}
|
|
|
|
data, err := Download(feed.Url, feed.Proxy)
|
|
if err != nil {
|
|
log.Error("download failed", "feed", feed.Name, "url", feed.Url, "proxy", feed.Proxy, "error", err)
|
|
return
|
|
}
|
|
|
|
entries, err := parser.ParseFeed(feed.Type, data)
|
|
if err != nil {
|
|
log.Error("feed parsing failed", "feed", feed.Name, "data", data, "error", err)
|
|
return
|
|
}
|
|
|
|
scheduler.Good(feed.Name)
|
|
|
|
var records []Record
|
|
var newEntries []parser.FeedEntry
|
|
if db.IsNewFeed() {
|
|
ftl := int(feed.FirstTimeLimit)
|
|
if feed.FirstTimeLimit == NoLimit || len(entries) <= ftl {
|
|
newEntries = entries
|
|
} else {
|
|
var filteredEntries []parser.FeedEntry
|
|
newEntries, filteredEntries = entries[:ftl], entries[ftl:]
|
|
for _, entry := range filteredEntries {
|
|
records = append(records, Record{
|
|
Time: time.Now(),
|
|
Status: Filtered,
|
|
Filter: "FirstTime",
|
|
FeedEntry: entry,
|
|
})
|
|
}
|
|
}
|
|
} else {
|
|
newEntries, err = db.Filter(entries)
|
|
if err != nil {
|
|
log.Error("failed to filter entries", "feed", feed.Name, "error", err)
|
|
}
|
|
}
|
|
|
|
for _, entry := range newEntries {
|
|
r := Record{
|
|
Time: time.Now(),
|
|
FeedEntry: entry,
|
|
}
|
|
err := tgram.SendLink(entry.Link, feed.Channel, feed.Rhash)
|
|
if err != nil {
|
|
log.Error("failed to send to telegram", "feed", feed.Name, "link", entry.Link, "channel", feed.Channel, "rhash", feed.Rhash, "error", err)
|
|
r.Status = Error
|
|
} else {
|
|
r.Status = Sent
|
|
}
|
|
records = append(records, r)
|
|
}
|
|
err = db.Save(records)
|
|
if err != nil {
|
|
log.Error("failed to save sent records", "feed", feed.Name, "num_records", len(records), "error", err)
|
|
}
|
|
}
|
|
|
|
func Download(url string, proxyUrl string) ([]byte, error) {
|
|
transport, err := proxy.GetTransport(proxyUrl)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
client := &http.Client{Transport: transport}
|
|
res, err := client.Get(url)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer res.Body.Close()
|
|
return io.ReadAll(res.Body)
|
|
}
|