dracoblue.net

def _parse_gs_uri(gcs_uri: str) -> tuple[str, str]:
    if not gcs_uri.startswith("gs://"):
        raise ValueError(f"Expected gs:// URI, got: {gcs_uri}")
    without = gcs_uri[len("gs://"):]
    bucket, _, prefix = without.partition("/")
    return bucket, prefix.strip("/")

def download_model_from_gcs(gcs_uri: str, project_id: Optional[str]) -> str:
    bucket, prefix = _parse_gs_uri(gcs_uri)
    client_kwargs = {"project": project_id} if project_id else {}
    gcs = storage.Client(**client_kwargs)
    bkt = gcs.bucket(bucket)

    tmpdir = pathlib.Path(tempfile.mkdtemp(prefix="bertopic_model_"))

    local_file = tmpdir / os.path.basename(prefix)
    bkt.blob(prefix).download_to_filename(str(local_file))
    return str(local_file)

def upload_model_to_gcs(local_model_file: str, gcs_uri: str, project_id: Optional[str]):
    bucket, prefix = _parse_gs_uri(gcs_uri)
    client_kwargs = {"project": project_id} if project_id else {}
    gcs = storage.Client(**client_kwargs)
    bkt = gcs.bucket(bucket)
    b = bkt.blob(prefix)
    b.upload_from_filename(local_model_file, timeout=600)

def generate_missing_embeddings_for_data(bq: bigquery.Client, project: str, dataset: str, emb_table: str, batch_date: str, embedding_model: str, transcripts_table: str, transcripts_cleaned_table: str):
    sql = f"""

    INSERT INTO `{project}.{dataset}.{emb_table}` (
        video_youtube_id, text, text_clean, emb, batch_date
    )
    SELECT
        ge.video_youtube_id,
        t.text AS text,
        ge.content AS text_clean,
        ge.ml_generate_embedding_result AS emb,
        DATE(t.published_at) AS batch_date
    FROM ML.GENERATE_EMBEDDING(
        MODEL `{project}.{embedding_model}`,  
        (
            SELECT
                t.video_youtube_id,
                s.text_clean AS content               
            FROM `{project}.{dataset}.{transcripts_table}` t
            JOIN `{project}.{dataset}.{transcripts_cleaned_table}` s
                USING (video_youtube_id)
            LEFT JOIN `{project}.{dataset}.{emb_table}` ve
                ON ve.video_youtube_id = t.video_youtube_id
            WHERE DATE(s.published_at) = DATE(@d)
                AND ve.video_youtube_id IS NULL
        ),
        STRUCT(TRUE AS flatten_json_output)      
    ) AS ge
    LEFT JOIN `{project}.{dataset}.{transcripts_table}` t
        ON t.video_youtube_id = ge.video_youtube_id
    ;
    """
    job = bq.query(sql, job_config=bigquery.QueryJobConfig(
        query_parameters=[bigquery.ScalarQueryParameter("d", "DATE", batch_date)]
    ))

    return job.result()

import umap, hdbscan
import numpy as np
from bertopic import BERTopic
import pathlib, tempfile

def build_bertopic(umap_neighbors, umap_components, min_cluster_size, min_samples, top_n_words) -> BERTopic:
    umap_model = umap.UMAP(
        n_neighbors=umap_neighbors, n_components=umap_components,
        min_dist=0.0, metric="cosine", random_state=42
    )
    hdbscan_model = hdbscan.HDBSCAN(
        min_cluster_size=min_cluster_size, min_samples=min_samples,
        metric="euclidean", cluster_selection_method="eom", prediction_data=True
    )
    return BERTopic(
        umap_model=umap_model, hdbscan_model=hdbscan_model,
        top_n_words=top_n_words, calculate_probabilities=True,
        language="multilingual", verbose=True, nr_topics=None
    )

def train_model(project_id, model_gcs_path, bq, dataset, emb_table,
                lookback_days, min_docs, umap_neighbors, umap_components, min_cluster_size, min_samples, top_n_words):
    train_df = read_embeddings_lookback(bq, project_id, dataset, emb_table, lookback_days)
    if train_df.empty or len(train_df) < min_docs:
        raise RuntimeError(f"Not enough docs to train initial model (have {len(train_df)}, need >= {min_docs}).")
    texts = train_df["text"].tolist()
    embs = np.vstack(train_df["emb"].apply(lambda a: np.array(a, dtype=np.float32)).to_numpy())
    topic_model = build_bertopic(umap_neighbors, umap_components, min_cluster_size, min_samples, top_n_words)
    topic_model.fit(texts, embeddings=embs)

    tmp = pathlib.Path(tempfile.mkdtemp(prefix="bertopic_save_")) / "model.pkl"
    topic_model.save(str(tmp))
    upload_model_to_gcs(str(tmp), model_gcs_path, project_id)
    return topic_model

    PROJECT_ID = required("PROJECT_ID")
    BQ_DATASET = os.getenv("BQ_DATASET", "topicwatchdog")
    EMB_TABLE = os.getenv("EMB_TABLE", "video_embeddings")
    VIDEO_TOPICS_TABLE = os.getenv("VIDEO_TOPICS_TABLE", "video_topics")
    DIM_TOPICS_TABLE = os.getenv("DIM_TOPICS_TABLE", "dim_topics")
    TOPIC_REGISTRY_TABLE = os.getenv("TOPIC_REGISTRY_TABLE", "topic_registry")
    MODEL_VERSION = required("MODEL_VERSION")
    MODEL_GCS_BASE_PATH = required("MODEL_GCS_BASE_PATH")
    MODEL_GCS_PATH = f"{MODEL_GCS_BASE_PATH}/model_{MODEL_VERSION}.pkl"
    BATCH_DATE = os.getenv("BATCH_DATE")
    JOB_MODE = os.getenv("JOB_MODE", "INFERENCE")

    EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text_embedding")
    TRANSCRIPTS_TABLE = os.getenv("TRANSCRIPTS_TABLE", "transcripts")
    TRANSCRIPTS_CLEANED_TABLE = os.getenv("TRANSCRIPTS_CLEANED_TABLE", "transcripts_cleaned")

    # Initial-train knobs
    LOOKBACK_DAYS = as_int("INIT_TRAIN_LOOKBACK_DAYS", 30)
    MIN_DOCS = as_int("INIT_MIN_DOCS", 200)
    UMAP_N_NEIGHBORS = as_int("UMAP_N_NEIGHBORS", 15)
    UMAP_N_COMPONENTS = as_int("UMAP_N_COMPONENTS", 5)
    MIN_CLUSTER_SIZE = as_int("MIN_CLUSTER_SIZE", 15)  
    MIN_SAMPLES = as_int("MIN_SAMPLES", 5) 
    TOP_N_WORDS = as_int("TOP_N_WORDS", 10)

    bq = bq_client(PROJECT_ID)

    if JOB_MODE.upper() == "TRAIN":
        # Load or initial-train the model
        topic_model = train_model(
            project_id=PROJECT_ID,
            model_gcs_path=MODEL_GCS_PATH,
            bq=bq,
            dataset=BQ_DATASET,
            emb_table=EMB_TABLE,
            lookback_days=LOOKBACK_DAYS,
            min_docs=MIN_DOCS,
            umap_neighbors=UMAP_N_NEIGHBORS,
            umap_components=UMAP_N_COMPONENTS,
            min_cluster_size=MIN_CLUSTER_SIZE,
            min_samples=MIN_SAMPLES,
            top_n_words=TOP_N_WORDS,
        )
        return 

    topic_model = load_model(
        project_id=PROJECT_ID,
        model_gcs_path=MODEL_GCS_PATH,
    )

    topic_info = topic_model.get_topic_info()

    track("calculateTopics", {
        "batchDate": BATCH_DATE,
        "modelVersion": MODEL_VERSION,
        "count": len(topic_info)
    })

    print(f"[INFO] Model {MODEL_VERSION} knows {len(topic_info)} topics")

    generate_missing_embeddings_for_data(bq, PROJECT_ID, BQ_DATASET, EMB_TABLE, BATCH_DATE, EMBEDDING_MODEL, TRANSCRIPTS_TABLE, TRANSCRIPTS_CLEANED_TABLE);

def read_video_embeddings_for_date(bq: bigquery.Client, project: str, dataset: str, emb_table: str, batch_date: str) -> pd.DataFrame:
    sql = f"""
    SELECT video_youtube_id, text, emb, batch_date
    FROM `{project}.{dataset}.{emb_table}`
    WHERE batch_date = DATE(@d)
    """
    job = bq.query(sql, job_config=bigquery.QueryJobConfig(
        query_parameters=[bigquery.ScalarQueryParameter("d", "DATE", batch_date)]
    ))
    return job.result().to_dataframe(progress_bar_type=None)

    df = read_video_embeddings_for_date(bq, PROJECT_ID, BQ_DATASET, EMB_TABLE, BATCH_DATE)
    if df.empty:
        print(f"[INFO] No embeddings for {BATCH_DATE}. Exiting.")
        return

    texts = df["text"].tolist()
    embs = np.vstack(df["emb"].apply(lambda a: np.array(a, dtype=np.float32)).to_numpy())
    topics, probs = topic_model.transform(texts, embeddings=embs)

    outlier_rate = (np.array(topics) == -1).mean()
    print(f"[INFO] Outlier-Rate for {BATCH_DATE}: {outlier_rate:.2%}")

    df["internal_topic_id"] = [int(t) for t in topics]
    df["confidence"] = [float(p) if p is not None else None for p in probs.max(axis=1)]
    df["batch_date"] = pd.to_datetime(BATCH_DATE).date()

    # Registry & dim_topics bootstrap/mapping
    registry = load_topic_registry(bq, PROJECT_ID, BQ_DATASET, TOPIC_REGISTRY_TABLE, MODEL_VERSION)
    known = set(registry.keys())
    seen = set(df["internal_topic_id"].unique().tolist())
    new_internal = sorted(list(seen - known))
    next_id_start = get_max_topic_id(bq, PROJECT_ID, BQ_DATASET, DIM_TOPICS_TABLE) + 1

    # Labels/keywords and centroids for new internal topics
    topic_labels: Dict[int, str] = {}
    topic_keywords: Dict[int, List[str]] = {}
    for internal_id in new_internal:
        if internal_id == -1:
            topic_labels[internal_id] = "OUTLIER"
            topic_keywords[internal_id] = []
        else:
            top_words = topic_model.get_topic(internal_id) or []
            kws = [w for (w, _ctfidf) in top_words][:TOP_N_WORDS]
            topic_keywords[internal_id] = kws
            topic_labels[internal_id] = ", ".join(kws[:4]) if kws else f"topic_{internal_id}"

    # Centroids from this batch (good enough for bootstrap)
    centroids: Dict[int, List[float]] = {}
    for internal_id in new_internal:
        mask = df["internal_topic_id"] == internal_id
        if mask.any():
            c = embs[mask].mean(axis=0)
            centroids[internal_id] = [float(x) for x in c.tolist()]
        else:
            centroids[internal_id] = []

    new_map_rows = []
    for i, internal_id in enumerate(new_internal):
        stable_id = next_id_start + i if internal_id != -1 else -1
        confs = df.loc[df["internal_topic_id"] == internal_id, "confidence"]
        new_map_rows.append({
            "internal_topic_id": internal_id,
            "topic_id": stable_id,
            "confidence_min": float(confs.min()) if len(confs) else None,
            "confidence_mean": float(confs.mean()) if len(confs) else None
        })

    if new_map_rows:
        upsert_new_registry_and_dim_topics(
            bq=bq,
            project=PROJECT_ID,
            dataset=BQ_DATASET,
            registry_table=TOPIC_REGISTRY_TABLE,
            dim_topics_table=DIM_TOPICS_TABLE,
            model_version=MODEL_VERSION,
            new_mappings=new_map_rows,
            topic_labels=topic_labels,
            topic_keywords=topic_keywords,
            batch_centroids=centroids
        )
        registry = load_topic_registry(bq, PROJECT_ID, BQ_DATASET, TOPIC_REGISTRY_TABLE, MODEL_VERSION)

    df["topic_id"] = df["internal_topic_id"].apply(lambda t: registry.get(int(t), -1))

    # Join readable labels/keywords
    label_map, keywords_map = {}, {}
    for r in bq.query(f"""
        SELECT topic_id, label, keywords
        FROM `{PROJECT_ID}.{BQ_DATASET}.{DIM_TOPICS_TABLE}`
        WHERE active_to IS NULL
    """).result():
        label_map[int(r["topic_id"])] = r["label"]
        keywords_map[int(r["topic_id"])] = r["keywords"]

    df["topic"] = df["topic_id"].map(lambda t: label_map.get(int(t), "UNKNOWN"))
    df["description"] = df["topic_id"].map(lambda t: ", ".join(keywords_map.get(int(t), []) or []))

    # video_topics (snippet_id = video id; no timestamp on whole-video embedding)
    df["topic_timestamp"] = None

    out = df[["video_youtube_id", "topic_id", "topic", "description", "confidence", "batch_date"]].copy()
    upsert_new_video_topics(bq, PROJECT_ID, BQ_DATASET, VIDEO_TOPICS_TABLE, out)

-- make sure all labels are empty
UPDATE topicwatchdog.topic_registry SET topic_label = NULL WHERE true;

-- fetch some clusters examples as input to gemini model
CREATE TEMP TABLE _cluster_examples AS
SELECT
  vt.topic_id,
  ARRAY_AGG(text_clean ORDER BY dv.views_count DESC LIMIT 50) AS examples,
  COUNT(*) AS n
FROM `topicwatchdog.video_topics` vt
LEFT JOIN topicwatchdog.topic_registry tr ON vt.topic_id = tr.topic_id
LEFT JOIN topicwatchdog.transcripts_cleaned tc ON vt.video_youtube_id = tc.video_youtube_id
LEFT JOIN topicwatchdog.dim_videos dv ON dv.video_youtube_id = vt.video_youtube_id
WHERE topic_label IS NULL OR topic_label = ''
GROUP BY vt.topic_id;

-- store the prompts by using these examples
CREATE TEMP TABLE _prompts AS
SELECT
  topic_id,
  (
    'Du bist politischer Analyst. Vergib einen prägnanten, verständlichen ' ||
    'Titel (max. 120 Zeichen) für diesen **politischen Anspruch/Behauptung**-Cluster. ' ||
    'Regeln: deutsch, neutral, keine neuen Themen/Namen erfinden, ' ||
    'nur EIN kurzer Titel. Antworte strikt Text ohne Markdown oder json. ' ||
    '\n\nBeispiele (Original-Canonicals):\n' ||
    STRING_AGG('- ' || ex, '\n')
  ) AS prompt
FROM _cluster_examples, UNNEST(examples) AS ex
GROUP BY topic_id;

-- get all the topic_label entries for topic_id by executing the prompts
CREATE TEMP TABLE _raw_llm AS
SELECT
  topic_id,
  ml_generate_text_llm_result AS topic_label
FROM ML.GENERATE_TEXT(
  MODEL `topicwatchdog.gemini_remote`,
  TABLE _prompts,
  STRUCT(
    TRUE AS flatten_json_output,
    0.2 AS temperature,
    1024 AS max_output_tokens
  )
) AS gen;

-- store them into the topic_registry as new topic_label
MERGE `topicwatchdog.topic_registry` dst
USING (
  SELECT topic_id, topic_label
  FROM _raw_llm
  WHERE topic_label IS NOT NULL
) src
ON dst.topic_id = src.topic_id
WHEN MATCHED THEN
  UPDATE SET dst.topic_label = src.topic_label

| topic_id | cluster_label                                              |
| -------- | ---------------------------------------------------------- |
| 1        | Politische Abgrenzung und Ablehnung                        |
| 2        | Abgrenzung und Ablehnung politischer Strömungen            |
| …        | …                                                          |