Import missing RRSS months from live Drive calendars.
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>
This commit is contained in:
4Ventures
569
scripts/import_drive_calendars_live.py
Normal file
569
scripts/import_drive_calendars_live.py
Normal file
@@ -0,0 +1,569 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user