feef339c2b
Agua Miia and TYD had many tabs in Drive that were never in documents; add a live importer and expand the batch JOBS map so future syncs cover them. Co-authored-by: Cursor <cursoragent@cursor.com>
570 lines
17 KiB
Python
570 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Import RRSS calendars from live Google Drive sheets → documents (type=calendar).
|
||
|
||
python3 scripts/import_drive_calendars_live.py
|
||
|
||
Exports each client's CALENDARIO spreadsheet, maps tabs → YYYY-MM via weekday
|
||
alignment, parses SEMANA×STORIES/POST/LINKEDIN grids, upserts missing/thin months.
|
||
Preserves richer existing plans (e.g. Agua Miia Oct 2026 bundled) when they have
|
||
more pieces than the Drive grid parse.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import urllib.parse
|
||
import urllib.request
|
||
import zipfile
|
||
import io
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from openpyxl import load_workbook
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
ENV = ROOT / ".env.local"
|
||
OUT = Path("/tmp/drive-cals")
|
||
|
||
DRIVE_SHEETS = {
|
||
"agua-miia": {
|
||
"brand": "Agua Miia",
|
||
"file_id": "18Dl4ln6uPkoSHK2CABdjYIQYbB870gsxdHtgGDKW8Tw",
|
||
},
|
||
"bps": {
|
||
"brand": "Blue Parrot School",
|
||
"file_id": "1YPcHBjioiHHQy22Q2xevkvUqA5jMq87v6IviV7VTL2s",
|
||
},
|
||
"brava-dermoestetica": {
|
||
"brand": "Brava Dermoestética",
|
||
"file_id": "1blboVr4Ig3xb_YQjO84_I6_yDuT0MjkdKYWjbeNY4dc",
|
||
},
|
||
"halal-safety": {
|
||
"brand": "Halal Safety",
|
||
"file_id": "1SOZQMFFRulLeafd2ojCjBSykvPKRA6npJvOyp3b0_r4",
|
||
},
|
||
"noberplast": {
|
||
"brand": "Noberplast",
|
||
"file_id": "1GhDxBOem20Fv7nFWfJKN445S1ixe_7W4_f1WeGsRPdc",
|
||
# Single "Hoja 1" sheet — ops calendar for Oct 2026 (Jueves 1 also matches enero).
|
||
"force_sheet_months": {"Hoja 1": "2026-10"},
|
||
},
|
||
"talento-y-deporte": {
|
||
"brand": "Talento y Deporte",
|
||
"file_id": "1HSbJAfjlBDdpHJubgxhT322UUZVJ2zlQHaPkO0HtevA",
|
||
},
|
||
"4ventures-hub": {
|
||
"brand": "4Ventures",
|
||
"file_id": "1OJVh2FJzXI5-yRm9LB6InbnteUAHxqOT7u6Ch2tEv7Y",
|
||
},
|
||
}
|
||
|
||
MONTHS = {
|
||
"ENERO": 1,
|
||
"FEBRERO": 2,
|
||
"MARZO": 3,
|
||
"ABRIL": 4,
|
||
"MAYO": 5,
|
||
"JUNIO": 6,
|
||
"JULIO": 7,
|
||
"AGOSTO": 8,
|
||
"SEPTIEMBRE": 9,
|
||
"SEPTIMEBRE": 9,
|
||
"OCTUBRE": 10,
|
||
"NOVIEMBRE": 11,
|
||
"DICIEMBRE": 12,
|
||
}
|
||
WD = {
|
||
"LUNES": 0,
|
||
"MARTES": 1,
|
||
"MIERCOLES": 2,
|
||
"JUEVES": 3,
|
||
"JUVES": 3,
|
||
"VIERNES": 4,
|
||
"SABADO": 5,
|
||
"DOMINGO": 6,
|
||
}
|
||
SKIP_TABS = re.compile(
|
||
r"ideas|calendario$|publicidad|tareas|competidores|mayo-diciembre",
|
||
re.I,
|
||
)
|
||
# "Hoja N" only skipped when we cannot infer a month; single-sheet clients (Noberplast) keep them.
|
||
MONTH_LABELS = [
|
||
"",
|
||
"Enero",
|
||
"Febrero",
|
||
"Marzo",
|
||
"Abril",
|
||
"Mayo",
|
||
"Junio",
|
||
"Julio",
|
||
"Agosto",
|
||
"Septiembre",
|
||
"Octubre",
|
||
"Noviembre",
|
||
"Diciembre",
|
||
]
|
||
|
||
|
||
def load_env() -> dict[str, str]:
|
||
env: dict[str, str] = {}
|
||
for line in ENV.read_text().splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, v = line.split("=", 1)
|
||
env[k] = v.strip().strip('"').strip("'")
|
||
return env
|
||
|
||
|
||
def rest(env: dict[str, str], path: str, method: str = "GET", body: Any = None):
|
||
url = (env.get("NEXT_PUBLIC_SUPABASE_URL") or env.get("SUPABASE_URL") or "").rstrip(
|
||
"/"
|
||
)
|
||
key = env["SUPABASE_SERVICE_ROLE_KEY"]
|
||
headers = {
|
||
"apikey": key,
|
||
"Authorization": f"Bearer {key}",
|
||
"Content-Type": "application/json",
|
||
"Prefer": "return=representation",
|
||
}
|
||
data = None if body is None else json.dumps(body).encode()
|
||
req = urllib.request.Request(url + path, data=data, headers=headers, method=method)
|
||
with urllib.request.urlopen(req, timeout=120) as r:
|
||
raw = r.read().decode()
|
||
return json.loads(raw) if raw else None
|
||
|
||
|
||
def drive_token(env: dict[str, str]) -> str:
|
||
oauth = rest(
|
||
env, "/rest/v1/agency_oauth?provider=eq.google_drive&select=refresh_token"
|
||
)[0]
|
||
cid = env.get("GOOGLE_OAUTH_CLIENT_ID") or env.get("NEXT_PUBLIC_GOOGLE_CLIENT_ID")
|
||
csec = env.get("GOOGLE_OAUTH_CLIENT_SECRET")
|
||
req = urllib.request.Request(
|
||
"https://oauth2.googleapis.com/token",
|
||
data=urllib.parse.urlencode(
|
||
{
|
||
"client_id": cid,
|
||
"client_secret": csec,
|
||
"refresh_token": oauth["refresh_token"],
|
||
"grant_type": "refresh_token",
|
||
}
|
||
).encode(),
|
||
)
|
||
with urllib.request.urlopen(req, timeout=60) as r:
|
||
return json.loads(r.read().decode())["access_token"]
|
||
|
||
|
||
def export_xlsx(token: str, file_id: str, dest: Path) -> Path:
|
||
export_url = (
|
||
f"https://www.googleapis.com/drive/v3/files/{file_id}/export?"
|
||
+ urllib.parse.urlencode(
|
||
{
|
||
"mimeType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
}
|
||
)
|
||
)
|
||
req = urllib.request.Request(
|
||
export_url, headers={"Authorization": f"Bearer {token}"}
|
||
)
|
||
with urllib.request.urlopen(req, timeout=180) as r:
|
||
dest.write_bytes(r.read())
|
||
return dest
|
||
|
||
|
||
def slugify(s: str) -> str:
|
||
import unicodedata
|
||
|
||
s = unicodedata.normalize("NFD", s.lower())
|
||
s = "".join(c for c in s if unicodedata.category(c) != "Mn")
|
||
s = re.sub(r"[^a-z0-9]+", "-", s).strip("-")
|
||
return s[:48]
|
||
|
||
|
||
def infer_month_num(name: str) -> int | None:
|
||
u = name.upper().replace("SEPTIMEBRE", "SEPTIEMBRE")
|
||
for m, n in MONTHS.items():
|
||
if m in u:
|
||
return n
|
||
return None
|
||
|
||
|
||
def normalize_wd(name: str) -> str:
|
||
return (
|
||
name.upper()
|
||
.replace("É", "E")
|
||
.replace("Á", "A")
|
||
.replace("Í", "I")
|
||
.replace("Ó", "O")
|
||
.replace("Ú", "U")
|
||
)
|
||
|
||
|
||
def sample_headers(ws) -> list[tuple[str, int]]:
|
||
hits: list[tuple[str, int]] = []
|
||
for row in ws.iter_rows(min_row=1, max_row=50, max_col=12, values_only=True):
|
||
for v in row or []:
|
||
if v is None:
|
||
continue
|
||
s = normalize_wd(str(v).strip())
|
||
m = re.match(
|
||
r"(LUNES|MARTES|MIERCOLES|JUEVES|JUVES|VIERNES|SABADO|DOMINGO)\s+(\d{1,2})\s*$",
|
||
s,
|
||
)
|
||
if m:
|
||
hits.append((m.group(1), int(m.group(2))))
|
||
return hits[:20]
|
||
|
||
|
||
def best_year(month: int, hits: list[tuple[str, int]], tab_name: str) -> int | None:
|
||
# Explicit year in tab wins when present
|
||
ym = re.search(r"(20\d{2})", tab_name)
|
||
if ym:
|
||
return int(ym.group(1))
|
||
if re.search(r"\b26\b", tab_name) or tab_name.upper().endswith("26"):
|
||
return 2026
|
||
|
||
scores: dict[int, tuple[int, int]] = {}
|
||
for y in (2024, 2025, 2026, 2027):
|
||
ok = tot = 0
|
||
for wdname, day in hits:
|
||
if day < 1 or day > 31:
|
||
continue
|
||
try:
|
||
d = date(y, month, day)
|
||
except ValueError:
|
||
continue
|
||
want = WD.get(wdname)
|
||
if want is None:
|
||
continue
|
||
tot += 1
|
||
if d.weekday() == want:
|
||
ok += 1
|
||
scores[y] = (ok, tot)
|
||
if not any(v[1] for v in scores.values()):
|
||
return None
|
||
best = max(scores.items(), key=lambda kv: (kv[1][0], -abs(kv[0] - 2026)))
|
||
if best[1][0] == 0:
|
||
return None
|
||
return best[0]
|
||
|
||
|
||
def parse_day_header(h: str) -> int | None:
|
||
raw = str(h or "").strip()
|
||
if not raw or re.match(r"^x\b", raw, re.I) or re.match(r"^columna", raw, re.I):
|
||
return None
|
||
m = re.search(r"(\d{1,2})\s*$", raw)
|
||
if not m:
|
||
return None
|
||
n = int(m.group(1))
|
||
return n if 1 <= n <= 31 else None
|
||
|
||
|
||
def detect_kind(c0: str) -> str | None:
|
||
u = c0.upper()
|
||
if u.startswith("STORIES") or u == "STORY" or u.startswith("STORY "):
|
||
return "STORIES"
|
||
if u.startswith("LINKEDIN"):
|
||
return "LINKEDIN"
|
||
if u.startswith("POST") or u.startswith("REEL") or u.startswith("CARRUSEL"):
|
||
return "POST"
|
||
if u == "STORY" or u.startswith("HISTORIA"):
|
||
return "STORIES"
|
||
return None
|
||
|
||
|
||
def piece_meta(kind: str) -> dict[str, str]:
|
||
if kind == "STORIES":
|
||
return {
|
||
"platform": "instagram",
|
||
"format": "Stories",
|
||
"pillar": "stories",
|
||
"kind": "story",
|
||
"color": "#D9821F",
|
||
"suffix": "st",
|
||
}
|
||
if kind == "LINKEDIN":
|
||
return {
|
||
"platform": "linkedin",
|
||
"format": "Post",
|
||
"pillar": "linkedin",
|
||
"kind": "post",
|
||
"color": "#7C66B8",
|
||
"suffix": "li",
|
||
}
|
||
return {
|
||
"platform": "instagram",
|
||
"format": "Post",
|
||
"pillar": "feed",
|
||
"kind": "post",
|
||
"color": "#2071cf",
|
||
"suffix": "post",
|
||
}
|
||
|
||
|
||
def parse_week_grid(prefix: str, rows: list[list[Any]], year: int, month: int) -> list[dict]:
|
||
pieces: list[dict] = []
|
||
day_cols: list[tuple[int, int]] = []
|
||
kind: str | None = None
|
||
week_wraps = False
|
||
|
||
for row in rows:
|
||
if not row or not any(c is not None and str(c).strip() for c in row):
|
||
continue
|
||
c0 = str(row[0] or "").strip()
|
||
c0u = c0.upper()
|
||
|
||
if c0u.startswith("SEMANA") or re.match(
|
||
r"^(LUNES|MARTES|MI[EÉ]RCOLES)", str(row[1] if len(row) > 1 else "")
|
||
):
|
||
day_cols = []
|
||
for col in range(1, min(len(row), 12)):
|
||
day = parse_day_header(str(row[col] or ""))
|
||
if day:
|
||
day_cols.append((col, day))
|
||
week_wraps = any(
|
||
i > 0
|
||
and day_cols[i][1] < day_cols[i - 1][1]
|
||
and day_cols[i - 1][1] >= 28
|
||
and day_cols[i][1] <= 7
|
||
for i in range(len(day_cols))
|
||
)
|
||
kind = None
|
||
continue
|
||
|
||
if (
|
||
re.match(r"^(PUBLICADO|CLIENTE|APROBADO)$", c0u)
|
||
or c0u.startswith("GUIONES")
|
||
or c0u.startswith("CARPETA")
|
||
):
|
||
continue
|
||
|
||
detected = detect_kind(c0u)
|
||
if detected:
|
||
kind = detected
|
||
elif not kind:
|
||
continue
|
||
if not day_cols:
|
||
continue
|
||
|
||
meta = piece_meta(kind)
|
||
for col, day in day_cols:
|
||
if col >= len(row):
|
||
continue
|
||
raw = row[col]
|
||
if raw is None:
|
||
continue
|
||
text = str(raw).strip()
|
||
if (
|
||
not text
|
||
or text == "X"
|
||
or re.match(r"^columna", text, re.I)
|
||
or re.match(r"^(PUBLICADO|APROBADO)$", text, re.I)
|
||
):
|
||
continue
|
||
|
||
y, m = year, month
|
||
if week_wraps and day >= 28:
|
||
m = month - 1
|
||
if m < 1:
|
||
m = 12
|
||
y = year - 1
|
||
|
||
try:
|
||
dt = date(y, m, day)
|
||
except ValueError:
|
||
continue
|
||
|
||
date_s = dt.isoformat()
|
||
title = text.split("\n")[0][:120]
|
||
pid = f"{prefix}-{date_s}-{meta['suffix']}-{slugify(title)[:24]}"
|
||
pieces.append(
|
||
{
|
||
"id": pid,
|
||
"date": date_s,
|
||
"platform": meta["platform"],
|
||
"format": meta["format"],
|
||
"pillar": meta["pillar"],
|
||
"title": title,
|
||
"body": text,
|
||
"status": "planned",
|
||
"kind": meta["kind"],
|
||
"color": meta["color"],
|
||
}
|
||
)
|
||
|
||
seen: set[str] = set()
|
||
out = []
|
||
for p in pieces:
|
||
if p["id"] in seen:
|
||
continue
|
||
seen.add(p["id"])
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def sheet_rows(ws) -> list[list[Any]]:
|
||
rows = []
|
||
for row in ws.iter_rows(values_only=True):
|
||
rows.append(list(row))
|
||
return rows
|
||
|
||
|
||
def build_plan(brand: str, ym: str, pieces: list[dict], source: str) -> dict:
|
||
y, m = map(int, ym.split("-"))
|
||
label = f"{MONTH_LABELS[m]} {y}"
|
||
posts = sum(1 for p in pieces if p.get("kind") != "story")
|
||
stories = sum(1 for p in pieces if p.get("kind") == "story")
|
||
pieces = sorted(pieces, key=lambda p: p["date"])
|
||
return {
|
||
"kind": "rrss_month",
|
||
"month": ym,
|
||
"label": label,
|
||
"hero": {
|
||
"eyebrow": f"4VENTURES × {brand.upper()} · PLAN RRSS",
|
||
"title": f"Calendario editorial · {label}",
|
||
"lede": f"Importado desde Drive ({source}).",
|
||
"chips": [f"{posts} posts", f"{stories} stories", "Fuente: Google Drive"],
|
||
},
|
||
"pieces": pieces,
|
||
}
|
||
|
||
|
||
def existing_plans(env: dict[str, str], tenant_id: str) -> dict[str, dict]:
|
||
docs = rest(
|
||
env,
|
||
f"/rest/v1/documents?tenant_id=eq.{tenant_id}&type=eq.calendar&select=id,title,content",
|
||
)
|
||
out: dict[str, dict] = {}
|
||
for d in docs or []:
|
||
c = d.get("content")
|
||
if isinstance(c, str):
|
||
try:
|
||
c = json.loads(c)
|
||
except Exception:
|
||
continue
|
||
if isinstance(c, dict) and c.get("kind") == "rrss_month" and c.get("month"):
|
||
out[str(c["month"])] = {
|
||
"id": d["id"],
|
||
"title": d.get("title"),
|
||
"pieces": len(c.get("pieces") or []),
|
||
"content": c,
|
||
}
|
||
return out
|
||
|
||
|
||
def upsert_plan(env: dict[str, str], tenant_id: str, plan: dict) -> str:
|
||
title = f"RRSS {plan['label']}"
|
||
docs = rest(
|
||
env,
|
||
f"/rest/v1/documents?tenant_id=eq.{tenant_id}&type=eq.calendar&select=id,title,content",
|
||
)
|
||
for d in docs or []:
|
||
c = d.get("content")
|
||
if isinstance(c, str):
|
||
try:
|
||
c = json.loads(c)
|
||
except Exception:
|
||
c = None
|
||
month = c.get("month") if isinstance(c, dict) else None
|
||
same_title = d.get("title") == title or (
|
||
plan["label"].lower() in (d.get("title") or "").lower()
|
||
)
|
||
if month == plan["month"] or same_title:
|
||
rest(env, f"/rest/v1/documents?id=eq.{d['id']}", method="DELETE")
|
||
saved = rest(
|
||
env,
|
||
"/rest/v1/documents",
|
||
method="POST",
|
||
body={
|
||
"tenant_id": tenant_id,
|
||
"type": "calendar",
|
||
"title": title,
|
||
"content": plan,
|
||
"status": "approved",
|
||
},
|
||
)
|
||
return (saved or [{}])[0].get("id", "")
|
||
|
||
|
||
def resolve_tenant(env: dict[str, str], slug: str) -> str:
|
||
rows = rest(
|
||
env, f"/rest/v1/tenants?slug=eq.{urllib.parse.quote(slug)}&select=id&limit=1"
|
||
)
|
||
if not rows:
|
||
raise SystemExit(f"tenant not found: {slug}")
|
||
return rows[0]["id"]
|
||
|
||
|
||
def import_client(env: dict[str, str], token: str, slug: str, meta: dict) -> None:
|
||
brand = meta["brand"]
|
||
print(f"\n=== {slug} ({brand}) ===")
|
||
dest = OUT / f"{slug}.xlsx"
|
||
if not dest.exists() or dest.stat().st_size < 1000:
|
||
print(" exporting from Drive…")
|
||
export_xlsx(token, meta["file_id"], dest)
|
||
else:
|
||
print(f" using cached {dest}")
|
||
|
||
wb = load_workbook(dest, data_only=True)
|
||
tenant_id = resolve_tenant(env, slug)
|
||
existing = existing_plans(env, tenant_id)
|
||
prefix = re.sub(r"[^a-z0-9]+", "", slug)[:12] or "cal"
|
||
|
||
force_map = meta.get("force_sheet_months") or {}
|
||
for name in wb.sheetnames:
|
||
if SKIP_TABS.search(name or ""):
|
||
print(f" skip tab {name}")
|
||
continue
|
||
if re.match(r"^hoja\s*\d+$", name or "", re.I) and len(wb.sheetnames) > 1 and name not in force_map:
|
||
print(f" skip tab {name}")
|
||
continue
|
||
hits = sample_headers(wb[name])
|
||
if name in force_map:
|
||
ym = force_map[name]
|
||
year, month_n = map(int, ym.split("-"))
|
||
print(f" {name} (forced) → {ym}")
|
||
else:
|
||
month_n = infer_month_num(name)
|
||
if not month_n:
|
||
print(f" skip tab {name} (no month)")
|
||
continue
|
||
year = best_year(month_n, hits, name)
|
||
if not year:
|
||
print(f" skip tab {name} (cannot infer year)")
|
||
continue
|
||
ym = f"{year}-{month_n:02d}"
|
||
pieces = parse_week_grid(prefix, sheet_rows(wb[name]), year, month_n)
|
||
print(f" {name} → {ym} ({len(pieces)} pieces)")
|
||
if not pieces:
|
||
print(" SKIP empty")
|
||
continue
|
||
|
||
prev = existing.get(ym)
|
||
if prev and prev["pieces"] >= len(pieces) and prev["pieces"] >= 8:
|
||
# Keep richer portal plan (e.g. Agua Miia Oct operative JSON).
|
||
print(
|
||
f" KEEP existing ({prev['pieces']} pieces ≥ Drive {len(pieces)})"
|
||
)
|
||
continue
|
||
|
||
plan = build_plan(brand, ym, pieces, dest.name)
|
||
doc_id = upsert_plan(env, tenant_id, plan)
|
||
print(f" UPSERT {ym} → {doc_id[:8] if doc_id else '?'} ({len(pieces)} pieces)")
|
||
existing[ym] = {"id": doc_id, "title": plan["label"], "pieces": len(pieces), "content": plan}
|
||
|
||
|
||
def main() -> None:
|
||
env = load_env()
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
token = drive_token(env)
|
||
for slug, meta in DRIVE_SHEETS.items():
|
||
import_client(env, token, slug, meta)
|
||
print("\nDone.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|