feat: mazání data/ souborů starších než 30 dní
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
#!/usr/bin/env python3
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from datetime import datetime, timezone
|
||||
|
||||
def load_env(path=".env"):
|
||||
env = {}
|
||||
try:
|
||||
with open(path) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, _, val = line.partition("=")
|
||||
env[key.strip()] = val.strip().strip('"').strip("'")
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return env
|
||||
|
||||
def api_get(url, token):
|
||||
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
except urllib.error.HTTPError as e:
|
||||
print(f"HTTP {e.code} při volání {url}: {e.read().decode()}", file=sys.stderr)
|
||||
raise
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Chyba sítě při volání {url}: {e.reason}", file=sys.stderr)
|
||||
raise
|
||||
|
||||
def clean_content(content):
|
||||
text = re.sub(r'<a\b[^>]*class="[^"]*hashtag[^"]*"[^>]*>.*?</a>', "", content, flags=re.IGNORECASE)
|
||||
text = re.sub(r"<[^>]+>", " ", text)
|
||||
text = html.unescape(text)
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
def main():
|
||||
env = {**load_env(), **os.environ}
|
||||
|
||||
for var in ("NOVINKY_TOKEN", "INSTANCE_URL"):
|
||||
if not env.get(var):
|
||||
print(f"Chybí proměnná prostředí: {var}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
token = env["NOVINKY_TOKEN"]
|
||||
base_url = env["INSTANCE_URL"].rstrip("/")
|
||||
|
||||
try:
|
||||
statuses = api_get(f"{base_url}/api/v1/trends/statuses?limit=10", token)
|
||||
except Exception:
|
||||
sys.exit(1)
|
||||
|
||||
candidates = []
|
||||
for s in statuses:
|
||||
if "@" in s.get("account", {}).get("acct", ""):
|
||||
continue
|
||||
text = clean_content(s.get("content", ""))
|
||||
if len(text) < 10:
|
||||
continue
|
||||
reblogs = s.get("reblogs_count", 0)
|
||||
favourites = s.get("favourites_count", 0)
|
||||
candidates.append({
|
||||
"acct": s["account"]["acct"],
|
||||
"text": text,
|
||||
"reblogs": reblogs,
|
||||
"favourites": favourites,
|
||||
"score": reblogs + favourites,
|
||||
})
|
||||
|
||||
candidates.sort(key=lambda x: x["score"], reverse=True)
|
||||
top = candidates[:3]
|
||||
|
||||
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||||
os.makedirs("data", exist_ok=True)
|
||||
out_path = os.path.join("data", f"{today}.json")
|
||||
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
json.dump({"date": today, "top": top}, f, ensure_ascii=False, indent=2)
|
||||
|
||||
print(f"Uloženo: {out_path} ({len(top)} tootů)")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user