Files

4.7 KiB
Raw Permalink Blame History

extract-post-sources

1. sources/extract.py

import re
from urllib.parse import urlparse

_MAX_PER_POST = 5

# t.me/handle, t.me/s/handle, telegram.me/handle, @handle
_TG_RE = re.compile(
    r"(?:https?://)?(?:www\.)?(?:t\.me|telegram\.me)/(?:s/)?([a-zA-Z][a-zA-Z0-9_]{3,31})"
)
_AT_RE = re.compile(r"(?<!\w)@([a-zA-Z][a-zA-Z0-9_]{3,31})\b")
# внешние http(s)-ссылки (не t.me/telegram.me)
_URL_RE = re.compile(r"https?://([^\s<>\"']+)", re.I)


def _domain_of(url: str) -> str:
    return (urlparse(url if "://" in url else "https://" + url).netloc or "").lower()


def extract_sources_from_text(text: str, exclude_tg_url: str | None = None) -> list[dict]:
    """Ищет TG-каналы и внешние сайты в тексте. exclude_tg_url — ссылка на сам пост
    (t.me/channel/postid) — её отбрасываем, это не источник, а адрес поста."""
    out: list[dict] = []
    seen = set()

    def add(kind, address, slug_hint, domain):
        key = (kind, slug_hint or domain or address)
        if key in seen:
            return
        seen.add(key)
        out.append({"kind": kind, "address": address, "slug_hint": slug_hint, "domain": domain})

    text = (text or "")[:4000]
    for m in _TG_RE.finditer(text):
        ch = m.group(1)
        if exclude_tg_url and f"/{ch}/" in exclude_tg_url:
            continue  # ссылка на сам пост
        add("telegram", f"https://t.me/{ch}", ch, None)
    for m in _AT_RE.finditer(text):
        add("telegram", f"https://t.me/{m.group(1)}", m.group(1), None)
    for m in _URL_RE.finditer(text):
        u = m.group(1).rstrip(".,;:!?)]}\"'")
        dom = _domain_of(u)
        if not dom:
            continue
        if dom in ("t.me", "telegram.me") or dom.endswith(".t.me"):
            continue  # уже обработали выше
        add("site", u, None, dom)
        if len(out) >= _MAX_PER_POST:
            break
    return out[:_MAX_PER_POST]

Ограничение _MAX_PER_POST=5: чтобы не захламлять карточку десятками ссылок.

2. web/app.py

from sources.sources import parse_address, add_or_update_source

def _post_sources(post: dict) -> list[dict]:
    """Список источников, найденных в тексте поста: + флаг known (уже в реестре)."""
    from sources.extract import extract_sources_from_text
    found = extract_sources_from_text(post.get("text") or "", post.get("url"))
    if not found:
        return []
    with db() as conn:
        slugs = {r[0] for r in conn.execute("SELECT slug FROM sources")}
    for f in found:
        f["known"] = f["slug_hint"] in slugs
    return found

В candidates() и selected(): post_sources=_post_sources(sel) if sel else [].

POST /sources/add-from-post:

@app.post("/sources/add-from-post")
def sources_add_from_post(request: Request, address: str = Form(...),
                          kind: str = Form(""), crawler: str = Form("telegram"),
                          src: str = Form("candidates"), selected: int = Form(0), ...):
    _require_auth(request)
    parsed = parse_address(address, crawler)
    if not parsed:
        return RedirectResponse(url=_cand_back(..., did="⚠️ Не удалось разобрать адрес"), ...)
    slug = parsed["slug"]
    with db() as conn:
        exists = conn.execute("SELECT 1 FROM sources WHERE slug=?", (slug,)).fetchone()
    if exists:
        return RedirectResponse(url=_cand_back(..., did=f"Источник {slug} уже в реестре"), ...)
    ok, res = add_or_update_source({"name": f"Из поста: {slug}", "crawler": crawler, **parsed})
    did = f"➕ Источник {res} добавлен" if ok else f"❌ {res}"
    return RedirectResponse(url=_cand_back(..., did=did), ...)

3. Шаблоны

В candidates.html и selected.html (под бейджами, перед деталями):

{% if post_sources %}
<div class="small text-muted mb-1">
  🌐 Источники из поста:
  {% for s in post_sources %}
    {% if s.known %}
      <span class="badge bg-light text-dark">@{{ s.slug_hint or s.domain }} ✓</span>
    {% else %}
      <form method="post" action="/sources/add-from-post" class="d-inline align-middle">
        {{ hid }}
        <input type="hidden" name="address" value="{{ s.address }}">
        <input type="hidden" name="crawler" value="{% if s.kind == 'telegram' %}telegram{% else %}rss{% endif %}">
        <button class="btn btn-sm btn-outline-secondary py-0" title="Добавить {{ s.address }} в источники">
         ➕ @{{ s.slug_hint or s.domain }}
        </button>
      </form>
    {% endif %}
  {% endfor %}
</div>
{% endif %}