"""Kaynak hesap izleyici — resmi yayıncı vb. hesaplardan alıntılık içerik. TELİF SINIRI ------------ Yayıncının gol videosunu indirip kendi hesabımıza yüklemek telif ihlalidir ve YAPILMAZ. Burada yapılan tek şey, tweet'in LİNKİNİ alıp üstüne kendi yorumumuzu yazarak ALINTI TWEET atmaktır. Alıntı X'in kendi paylaşım mekanizmasıdır: video kaynağın hesabından oynar, kaynak görünür kalır, hiçbir içerik çoğaltılmaz. Ayrıca metinleri de kopyalamayız — LLM olayı kendi cümleleriyle yorumlar. """ import logging import re from datetime import datetime, timedelta from email.utils import parsedate_to_datetime from sqlalchemy.orm import Session from app.db import models, crud from app.x_engine import client as xclient, importance log = logging.getLogger("x_engine.sources") MAX_AGE_MINUTES = 90 # bu yaştan eski kaynak tweet alıntılanmaz FETCH_COUNT = 20 # ------------------------------------------------------------------ okuma def _walk_tweets(obj, out: list): """GraphQL yanıt ağacından tweet kayıtlarını topla. twikit'in Tweet modeli X'in alan değişikliği yüzünden kırık (KeyError: 'urls'), bu yüzden ham yanıt üzerinden gidilir. """ if isinstance(obj, dict): lg = obj.get("legacy") if isinstance(lg, dict) and lg.get("full_text") and obj.get("rest_id"): ent = lg.get("entities") or {} ext = lg.get("extended_entities") or {} media_items = (ext.get("media") or ent.get("media") or []) types = [m.get("type") for m in media_items] # Fotoğraflı gönderilerde görselin doğrudan adresi (pbs.twimg.com): # link kart oluşturduğu için görselin kendisi paylaşılır. photo_url = next( (m.get("media_url_https") for m in media_items if m.get("type") == "photo" and m.get("media_url_https")), "", ) out.append({ "id": str(obj["rest_id"]), "text": lg["full_text"], "created": lg.get("created_at"), "media": "video" if "video" in types else ( "photo" if "photo" in types else ""), "media_url": photo_url, "retweets": lg.get("retweet_count") or 0, "likes": lg.get("favorite_count") or 0, "is_retweet": lg.get("full_text", "").startswith("RT @"), }) for v in obj.values(): _walk_tweets(v, out) elif isinstance(obj, list): for v in obj: _walk_tweets(v, out) async def _fetch(client, handle: str, user_id: str | None): if not user_id: r, _ = await client.gql.user_by_screen_name(handle) user_id = r["data"]["user"]["result"]["rest_id"] r2, _ = await client.gql.user_tweets(user_id, FETCH_COUNT, None) found = [] _walk_tweets(r2, found) # tekilleştir, sırayı koru seen, uniq = set(), [] for t in found: if t["id"] in seen: continue seen.add(t["id"]) uniq.append(t) return user_id, uniq def _parse_time(raw: str): if not raw: return None try: dt = parsedate_to_datetime(raw) return dt.replace(tzinfo=None) - (dt.utcoffset() or timedelta()) except Exception: return None # ------------------------------------------------------------------ sınıflama # --- SAHA AKSİYONU (paylaşmak istediğimiz) --- GOAL_WORDS = ["gol", "golü", "golu", "ağlara", "aglara", "fileleri", "kaydediyor", "ağları", "aglari", "penaltı golü", "kafa vuruşu", "şık vuruş"] ACTION_WORDS = ["kurtarış", "kurtaris", "gole izin vermedi", "direkten", "kırmızı kart", "kirmizi kart", "penaltı", "penalti", "var incelemesi", "asist", "çalım", "calim"] # --- YÖNLENDİRME / TANITIM (paylaşmak İSTEMEDİĞİMİZ) --- # Yayıncının kendi sitesine/uygulamasına trafik çeken gönderiler: # "🔗 İzlemek için: ..." , "📲 GENİŞ ÖZET BURADA" gibi. Görselleri sadece # kapak/afiş; bizim paylaşacağımız bir olay anlatmıyorlar. PROMO_WORDS = ["izlemek için", "izlemek icin", "detaylar için", "detaylar icin", "tıklayın", "tiklayin", "tıkla", "hemen izle", "linkte", "profilimizde", "uygulamada", "🔗", "beinsports.com", "bein.link", "beinconnect", "tamamı için", "tamami icin", "devamı için", "devami icin", "haberin detayları"] # beIN'in tanıtım şablonu başlığı büyük harfle biter: "... ÖZET BURADA" PROMO_UPPER = ["BURADA", "TIKLA"] # --- STÜDYO / YORUM (paylaşmak İSTEMEDİĞİMİZ) --- # Yayıncının stüdyo programları, yorumcu değerlendirmeleri: saha görüntüsü # değil, konuşan kafa. Ajans hesabı bunları paylaşmaz. STUDIO_WORDS = ["💬", "yorumcu", "değerlendirme", "degerlendirme", "yorumları", "yorumlari", "programında", "programinda", "stüdyo", "studyo", "masası", "masasi", "analiz", "konuk", "sunucu"] INTERVIEW_WORDS = ["🎙️", "🎙", "açıklamaları", "aciklamalari", "basın toplantısı", "basin toplantisi", "röportaj", "roportaj"] SUMMARY_WORDS = ["özet", "ozet", "maç sonucu", "mac sonucu", "maçın özeti"] def classify(text: str) -> str: """İçerik türü. Yalnızca AUTO_KINDS otomatik paylaşılır.""" t = text.lower() # Tanıtım/yönlendirme EN ÖNCE: bu gönderilerde de "gol", "özet" geçiyor # ("⚽️ İŞTE GALİBİYET GOLÜ ... 🔗 Detaylar için:") ama içerik reklam. if any(w in t for w in PROMO_WORDS) or any(w in text for w in PROMO_UPPER): return "promo" # Stüdyo/röportaj: bu tweetlerde de "gol" kelimesi geçebiliyor if any(w in t for w in STUDIO_WORDS): return "studio" if any(w in t for w in INTERVIEW_WORDS): return "interview" if any(w in t for w in SUMMARY_WORDS): return "summary" if any(w in t for w in GOAL_WORDS): return "goal" if any(w in t for w in ACTION_WORDS): return "action" return "other" # Otomatik alıntılanabilecek türler: yalnızca saha aksiyonu. # Stüdyo yorumu, röportaj ve belirsiz içerik elle onaya kalır. AUTO_KINDS = ("goal", "action") def _norm_team(name: str) -> str: t = name.lower() for suf in (" jk", " fk", " sk", " a.ş.", " as", " sportif faaliyetler"): t = t.replace(suf, "") return t.strip() _TR_MAP = str.maketrans("ÇĞİIÖŞÜçğıiöşü", "CGIIOSUCGIIOSU") def _ascii_up(s: str) -> str: return s.translate(_TR_MAP).upper() # beIN maç etiketi: #BJKvEYP, #AMDvERZ ... _TAG_RE = re.compile(r"#([A-Za-zÇĞİIÖŞÜçğıiöşü]{2,4})[vV]([A-Za-zÇĞİIÖŞÜçğıiöşü]{2,4})\b") def _code_fits(code: str, team: str) -> bool: """Kısaltma takım adının harf sırasına uyuyor mu? BJK ⊂ BESIKTAS JK, EYP ⊂ EYUPSPOR, AMD ⊂ AMED — hepsi alt dizi. Tek başına gevşek bir eşleşme; bu yüzden yalnızca etiketteki İKİ kod da aynı maçın iki takımına uyduğunda kabul edilir. """ c, t = _ascii_up(code), _ascii_up(team).replace(" ", "") i = 0 for ch in t: if i < len(c) and ch == c[i]: i += 1 return i == len(c) def extract_match_tag(text: str) -> str | None: """Metindeki maç etiketini döner (#BJKvEYP). Yoksa None.""" m = _TAG_RE.search(text) return f"#{m.group(1)}v{m.group(2)}" if m else None def learn_hashtag(db: Session, match: models.Match, text: str) -> bool: """Yayıncının maç etiketini öğrenip maça kaydeder. Kendi tweetlerimizde bu etiketi kullanınca o maçın konuşmasına dahil oluruz. Etiket tahmin edilemediği için (kodlar kurala uymuyor) yalnızca kaynaktan öğrenilir. """ if not match or match.hashtag: return False tag = extract_match_tag(text) if not tag: return False match.hashtag = tag crud.add_log(db, f"Maç etiketi öğrenildi: {match.home_team}-{match.away_team} → {tag}", source="sources") return True def _match_by_tag(text: str, candidates) -> models.Match | None: for m in _TAG_RE.finditer(text): a, b = m.group(1), m.group(2) for mt in candidates: if not mt.home_team or not mt.away_team: continue if ((_code_fits(a, mt.home_team) and _code_fits(b, mt.away_team)) or (_code_fits(b, mt.home_team) and _code_fits(a, mt.away_team))): return mt return None def match_to_fixture(db: Session, text: str) -> models.Match | None: """Tweet metnindeki takım adlarından canlı/yeni biten maçı bul. Pencere geniş tutulur (12 saat): yayıncı maç özetini, röportajı ve "işte o gol" gönderisini maç bittikten saatler sonra da paylaşabiliyor. Ayırt ediciliği takım adı sağlar, zaman değil. """ since = datetime.utcnow() - timedelta(hours=12) candidates = ( db.query(models.Match) .filter( models.Match.status.in_(["live", "ht", "finished"]), models.Match.kickoff >= since, ) .all() ) low = text.lower() best, best_hits = None, 0 for m in candidates: hits = 0 for team in (m.home_team, m.away_team): if not team: continue key = _norm_team(team) # takım adının ilk kelimesi genelde ayırt edici (Beşiktaş, Amed...) first = key.split()[0] if key.split() else key if len(first) >= 4 and first in low: hits += 1 if hits > best_hits: best, best_hits = m, hits if best_hits >= 1: return best # Takım adı geçmiyorsa beIN'in maç etiketini çöz (#BJKvEYP) return _match_by_tag(text, candidates) # ------------------------------------------------------------------ toplama def poll_sources(db: Session) -> int: """Etkin kaynak hesapları tara, yeni tweet'leri kaydet. Yeni sayısı döner.""" posting_account = ( db.query(models.Account) .filter(models.Account.is_active == True, # noqa: E712 models.Account.status == "ok") .first() ) if not posting_account: return 0 sources = (db.query(models.SourceAccount) .filter_by(enabled=True).all()) if not sources: return 0 client = xclient._build_client(posting_account) cutoff = datetime.utcnow() - timedelta(minutes=MAX_AGE_MINUTES) total_new = 0 for src in sources: try: user_id, tweets = xclient._run(_fetch(client, src.handle, src.user_id or None)) src.user_id = user_id src.last_checked = datetime.utcnow() src.last_error = "" except Exception as e: src.last_error = str(e)[:300] src.last_checked = datetime.utcnow() db.commit() crud.add_log(db, f"Kaynak okunamadı @{src.handle}: {e}", level="warn", source="sources") continue ids = [t["id"] for t in tweets] mevcut = { r.tweet_id: r for r in db.query(models.SourceTweet) .filter(models.SourceTweet.tweet_id.in_(ids)).all() } # Bekleyen adayların etkileşimini tazele: beIN gönderisi ilk # dakikalarda 0 RT ile geliyor, önemi sonradan belli oluyor. for t in tweets: row = mevcut.get(t["id"]) if row and not row.used: row.retweets = t.get("retweets", 0) row.likes = t.get("likes", 0) row.score = importance.static_score( row.kind, row.media, row.text, db.get(models.Match, row.match_id) if row.match_id else None, row.retweets, row.likes, ) existing = set(mevcut) for t in tweets: if t["id"] in existing or t["is_retweet"]: continue posted = _parse_time(t["created"]) if posted and posted < cutoff: continue # bayat, alıntılamaya değmez kind = classify(t["text"]) m = match_to_fixture(db, t["text"]) if m: learn_hashtag(db, m, t["text"]) db.add(models.SourceTweet( source_id=src.id, handle=src.handle, tweet_id=t["id"], text=t["text"][:1000], media=t["media"], media_url=t.get("media_url", ""), kind=kind, match_id=m.id if m else None, posted_at=posted, retweets=t.get("retweets", 0), likes=t.get("likes", 0), score=importance.static_score( kind, t["media"], t["text"], m, t.get("retweets", 0), t.get("likes", 0)), # Tanıtım gönderileri baştan elenir: elle onay listesini # doldurmalarının bir anlamı yok. skipped=(kind == "promo"), )) total_new += 1 db.commit() if total_new: crud.add_log(db, f"Kaynak taraması: {total_new} yeni tweet", source="sources") return total_new def pick_for_quote(db: Session, limit: int = 2) -> list[models.SourceTweet]: """Otomatik paylaşılacak kaynak tweet'leri seç. Sıkı üç şart: 1. Tür SAHA AKSİYONU olacak (gol/kurtarış/kart). Stüdyo yorumu ve röportaj otomatik gitmez — konuşan kafa videosu paylaşmıyoruz. 2. MEDYASI olacak: video (linkle paylaşılır) ya da görseli çekilebilen fotoğraf (görsel indirilip kendi tweetimize eklenir). Medyasız gönderinin paylaşılacak bir şeyi yok. 3. Taze olacak. """ cutoff = datetime.utcnow() - timedelta(minutes=MAX_AGE_MINUTES) rows = ( db.query(models.SourceTweet) .join(models.SourceAccount, models.SourceTweet.source_id == models.SourceAccount.id) .filter( models.SourceTweet.used == False, # noqa: E712 models.SourceTweet.skipped == False, # noqa: E712 (models.SourceTweet.media == "video") | ((models.SourceTweet.media == "photo") & (models.SourceTweet.media_url != "")), models.SourceTweet.kind.in_(AUTO_KINDS), models.SourceAccount.enabled == True, # noqa: E712 models.SourceAccount.auto_quote == True, # noqa: E712 ) .filter( (models.SourceTweet.posted_at == None) | # noqa: E711 (models.SourceTweet.posted_at >= cutoff) ) .all() ) # ÖNEM SIRASINA göre seç. Puanın işi eleme değil SIRALAMA: bir gol # her hâlükârda paylaşılmaya değer, ama aynı anda 5 gol düşerse önce # derbi golü gitsin, alt lig golü sıraya girsin. Bekleyen aday # kaybolmaz — sonraki turda yine değerlendirilir (tazelik puanı düşer, # yani sırası doğal olarak geriler ve çok bayatlarsa pencereden çıkar). now = datetime.utcnow() rows.sort(key=lambda r: -importance.total_score(r, now)) return rows[:limit]