spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   src/main.py (ka-youtube)4# Desc:   Chaînes YouTube ULTRA DÉTAILLÉES SANS clé API via le JSON ytInitialData5#         de la page /videos : abonnés, nb de vidéos, description, avatar,6#         BANNIÈRE, badge vérifié + vidéos récentes avec MINIATURES (vues, durée,7#         date relative) et agrégats — plus un 2e passage /about (vues totales de8#         la chaîne, date de création, pays, LIENS EXTERNES auto-déclarés).9#         Accepte @handle, handle nu, ou id de chaîne (UC…).10# ==============================================================================11from __future__ import annotations1213import asyncio14import json15import re16import urllib.parse1718from apify import Actor1920from .net import Fetcher2122_INITIAL_RE = re.compile(r"var ytInitialData\s*=\s*(\{.*?\});</script>", re.S)23_COUNT_RE = re.compile(r"([\d][\d\s  .,]*)\s*([KkMB]?)")242526def _url_for(handle: str) -> str:27    h = urllib.parse.unquote(handle).strip()28    if h.startswith("UC") and len(h) == 24:29        return f"https://www.youtube.com/channel/{h}/videos"30    if not h.startswith("@"):31        h = "@" + h32    return f"https://www.youtube.com/{urllib.parse.quote(h)}/videos"333435def parse_text_count(text: str) -> int | None:36    """« 1.23M subscribers », « 12 345 vues », « 4,5 k » → entier."""37    if not text:38        return None39    m = _COUNT_RE.search(str(text))40    if not m:41        return None42    num = re.sub(r"[\s  ]", "", m.group(1))43    unit = (m.group(2) or "").upper()44    if unit:45        num = num.replace(",", ".")46        if num.count(".") > 1:47            num = num.replace(".", "", num.count(".") - 1)48        try:49            val = float(num)50        except ValueError:51            return None52        return int(val * {"K": 1e3, "M": 1e6, "B": 1e9}[unit])53    num = re.sub(r"[.,]", "", num)54    return int(num) if num.isdigit() else None555657def _walk(node, key: str):58    """Premier objet portant `key` dans l'arbre ytInitialData."""59    if isinstance(node, dict):60        if key in node:61            return node[key]62        for v in node.values():63            hit = _walk(v, key)64            if hit is not None:65                return hit66    elif isinstance(node, list):67        for v in node:68            hit = _walk(v, key)69            if hit is not None:70                return hit71    return None727374def _texts(node) -> list[str]:75    """Toutes les chaînes `content`/`simpleText`/`text` sous un nœud."""76    out: list[str] = []7778    def rec(n):79        if isinstance(n, dict):80            for k in ("content", "simpleText", "text"):81                if isinstance(n.get(k), str):82                    out.append(n[k])83            for v in n.values():84                rec(v)85        elif isinstance(n, list):86            for v in n:87                rec(v)88    rec(node)89    return out909192def _lockup_video(lv: dict) -> dict | None:93    """Carte vidéo moderne (lockupViewModel) → dict vidéo standardisé."""94    if lv.get("contentType") != "LOCKUP_CONTENT_TYPE_VIDEO":95        return None96    vid = lv.get("contentId")97    if not vid:98        return None99    md = lv.get("metadata") or {}100    title = ((md.get("lockupMetadataViewModel") or {}).get("title") or {}) \101        .get("content")102    strings = _texts(md)103    views = next((parse_text_count(s) for s in strings104                  if "view" in s.lower() or "vue" in s.lower()), None)105    published = next((s for s in strings106                      if "ago" in s.lower() or "il y a" in s.lower()), None)107    overlays = _texts(lv.get("contentImage") or {})108    duration = next((s for s in overlays109                     if re.fullmatch(r"\d?\d:\d\d(:\d\d)?", s.strip())), None)110    return {111        "id": vid,112        "url": f"https://www.youtube.com/watch?v={vid}",113        "title": (title or "")[:200],114        "views": views,115        "published": published,116        "duration": duration,117        "thumbnail": f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg",118    }119120121def _collect_videos(data: dict, cap: int = 15) -> list[dict]:122    out: list[dict] = []123124    def rec(node):125        if len(out) >= cap:126            return127        if isinstance(node, dict):128            lv = node.get("lockupViewModel")129            if isinstance(lv, dict):130                v = _lockup_video(lv)131                if v:132                    out.append(v)133                    return134            vr = node.get("videoRenderer") or node.get("gridVideoRenderer")135            if vr and vr.get("videoId"):136                out.append({137                    "id": vr["videoId"],138                    "url": f"https://www.youtube.com/watch?v={vr['videoId']}",139                    "title": "".join(140                        r.get("text", "") for r in141                        (vr.get("title") or {}).get("runs") or [])[:200],142                    "views": parse_text_count(143                        (vr.get("viewCountText") or {}).get("simpleText")),144                    "published": ((vr.get("publishedTimeText") or {})145                                  .get("simpleText")),146                    "duration": ((vr.get("lengthText") or {})147                                 .get("simpleText")),148                    "thumbnail": (f"https://i.ytimg.com/vi/{vr['videoId']}"149                                  f"/hqdefault.jpg"),150                })151                return152            for v in node.values():153                rec(v)154        elif isinstance(node, list):155            for v in node:156                rec(v)157158    rec(data)159    return out160161162def _s(v):163    """ViewModel : champ tantôt chaîne nue, tantôt {content: "..."}."""164    return v.get("content") if isinstance(v, dict) else v165166167_REL_RE = re.compile(168    r"(\d+)\s*(hour|day|week|month|year|heure|jour|semaine|mois|an)", re.I)169_REL_DAYS = {"hour": 1 / 24, "heure": 1 / 24, "day": 1, "jour": 1,170             "week": 7, "semaine": 7, "month": 30.44, "mois": 30.44,171             "year": 365.25, "an": 365.25}172173174def _rel_days(published: str | None) -> float | None:175    """« 3 weeks ago » / « il y a 2 mois » → ancienneté approx. en jours."""176    m = _REL_RE.search(published or "")177    if not m:178        return None179    unit = m.group(2).lower()180    unit = next((k for k in _REL_DAYS if unit.startswith(k)), None)181    return float(m.group(1)) * _REL_DAYS[unit] if unit else None182183184def _cadence(vids: list[dict]) -> float | None:185    """Vidéos/mois estimées d'après les dates relatives (approximatif)."""186    ages = sorted(a for a in (_rel_days(v.get("published")) for v in vids)187                  if a is not None)188    if len(ages) < 3 or ages[-1] <= ages[0]:189        return None190    return round((len(ages) - 1) / ((ages[-1] - ages[0]) / 30.44), 2)191192193def _banner_of(data: dict) -> str | None:194    """Bannière de chaîne : nouvel en-tête (imageBannerViewModel) ou ancien."""195    b = _walk(data, "imageBannerViewModel")196    if isinstance(b, dict):197        srcs = (b.get("image") or {}).get("sources") or []198        if srcs and srcs[-1].get("url"):199            return srcs[-1]["url"]200    thumbs = (((_walk(data, "c4TabbedHeaderRenderer") or {}).get("banner")201               or {}).get("thumbnails")) or []202    return thumbs[-1].get("url") if thumbs else None203204205def _about_of(data: dict) -> dict:206    """Panneau « À propos » (/about) → vues totales, date, pays, liens."""207    acv = _walk(data, "aboutChannelViewModel") or {}208    if not acv:209        return {}210    links = []211    for entry in (acv.get("links") or []):212        lv = entry.get("channelExternalLinkViewModel") or {}213        url = _s(lv.get("link"))214        if url:215            links.append({"title": _s(lv.get("title")), "url": url})216    return {217        "total_views": parse_text_count(_s(acv.get("viewCountText"))),218        "joined_date": _s(acv.get("joinedDateText")),219        "country": _s(acv.get("country")),220        "external_links": links[:10],221    }222223224async def main() -> None:225    async with Actor:226        inp = await Actor.get_input() or {}227        fetch_about = bool(inp.get("fetchAbout", True))228        usernames = [u.strip() for u in (inp.get("usernames") or [])229                     if u and u.strip()]230        proxy = await Actor.create_proxy_configuration(231            actor_proxy_input=inp.get("proxyConfiguration"))232        fetcher = Fetcher(proxy, delay=float(inp.get("requestDelay") or 1))233        sem = asyncio.Semaphore(int(inp.get("concurrency") or 3))234235        async def one(u: str) -> None:236            async with sem:237                miss = {"kind": "profile", "platform": "youtube",238                        "username": u, "found": False}239                try:240                    resp = await fetcher.get(241                        _url_for(u), session_id=u,242                        headers={"Accept-Language": "en-US,en;q=0.9"})243                except Exception as exc:244                    await Actor.push_data({**miss, "error": str(exc)[:200]})245                    return246                m = _INITIAL_RE.search(resp.text or "")247                if resp.status_code != 200 or not m:248                    await Actor.push_data(249                        {**miss, "error": f"http_{resp.status_code}"})250                    return251                try:252                    data = json.loads(m.group(1))253                except Exception:254                    await Actor.push_data({**miss, "error": "bad_json"})255                    return256                meta = _walk(data, "channelMetadataRenderer") or {}257                if not meta.get("externalId"):258                    await Actor.push_data({**miss, "error": "no_channel"})259                    return260                # compteurs : le nouvel en-tête n'a plus subscriberCountText ;261                # on lit les chaînes « N subscribers » / « N videos » du header262                header = _walk(data, "pageHeaderViewModel") or {}263                header_txt = _texts(header) or _texts(264                    _walk(data, "c4TabbedHeaderRenderer") or {})265                sub_text = next((t for t in header_txt266                                 if "subscriber" in t.lower()267                                 or "abonné" in t.lower()), "")268                vid_text = next((t for t in header_txt269                                 if ("video" in t.lower()270                                     or "vidéo" in t.lower())271                                 and any(c.isdigit() for c in t)), "")272                if not sub_text:  # ancien en-tête273                    sub_text = ((_walk(data, "subscriberCountText") or {})274                                .get("simpleText") or "")275                subs = parse_text_count(sub_text)276                videos_count = parse_text_count(vid_text)277                vids = _collect_videos(data)278                views = [v["views"] for v in vids279                         if isinstance(v["views"], int)]280                text = resp.text or ""281                is_verified = True if (282                    '"BADGE_STYLE_TYPE_VERIFIED"' in text283                    or '"BADGE_STYLE_TYPE_VERIFIED_ARTIST"' in text284                    or '"OFFICIAL_ARTIST_BADGE"' in text) else None285                about: dict = {}286                if fetch_about:287                    try:288                        resp_a = await fetcher.get(289                            _url_for(u).replace("/videos", "/about"),290                            session_id=f"{u}a",291                            headers={"Accept-Language": "en-US,en;q=0.9"})292                        ma = _INITIAL_RE.search(resp_a.text or "")293                        if ma:294                            about = _about_of(json.loads(ma.group(1)))295                    except Exception as exc:296                        Actor.log.warning(f"/about {u} : {exc}")297                await Actor.push_data({298                    "kind": "profile",299                    "platform": "youtube",300                    "found": True,301                    "username": u,302                    "channel_id": meta.get("externalId"),303                    "full_name": meta.get("title"),304                    "biography": (meta.get("description") or "")[:1000],305                    "followers": subs,306                    "videos_count": videos_count,307                    "is_family_safe": meta.get("isFamilySafe"),308                    "is_verified": is_verified,309                    "keywords": (meta.get("keywords") or "")[:300] or None,310                    "country": about.get("country") or _walk(data, "country"),311                    "canonical_url": meta.get("vanityChannelUrl"),312                    "avatar": (((meta.get("avatar") or {}).get("thumbnails")313                                or [{}])[-1].get("url")),314                    "banner": _banner_of(data),315                    "total_views": about.get("total_views"),316                    "joined_date": about.get("joined_date"),317                    "external_links": about.get("external_links") or [],318                    "avg_views": (round(sum(views) / len(views))319                                  if views else None),320                    "videos_per_month": _cadence(vids),321                    "is_live_now": ('"style":"LIVE"' in text322                                    or '"label":"LIVE"' in text) or None,323                    "has_shorts": ('"title":"Shorts"' in text) or None,324                    "top_video": (max(325                        (v for v in vids if isinstance(v["views"], int)),326                        key=lambda v: v["views"], default=None)),327                    "last_video_published": (vids[0]["published"]328                                             if vids else None),329                    "recent_videos": vids,330                })331332        await asyncio.gather(*[one(u) for u in usernames])333        Actor.log.info(f"terminé : {len(usernames)} chaînes")334