chore: scripts Dokploy/BPS (cron secret, Ads MCC, sync y probes)
Incluye utilidades de ops para CRON_SECRET, login customer Google Ads, sync de seguimiento BPS y probes de conexiones. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
4Ventures
450
scripts/gap_report_connections.py
Normal file
450
scripts/gap_report_connections.py
Normal file
@@ -0,0 +1,450 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Exact gap report: visible assets vs tenant matches + name candidates."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
META = "https://graph.facebook.com/v21.0"
|
||||
|
||||
|
||||
def load_env() -> dict[str, str]:
|
||||
vals: dict[str, str] = {}
|
||||
for line in (ROOT / ".env.local").read_text(encoding="utf-8").splitlines():
|
||||
if not line.strip() or line.strip().startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
vals[k.strip()] = v.strip().strip('"').strip("'")
|
||||
return vals
|
||||
|
||||
|
||||
ENV = load_env()
|
||||
BASE = ENV["NEXT_PUBLIC_SUPABASE_URL"].rstrip("/")
|
||||
KEY = ENV["SUPABASE_SERVICE_ROLE_KEY"]
|
||||
SH = {"apikey": KEY, "Authorization": f"Bearer {KEY}"}
|
||||
|
||||
|
||||
def sb(path: str):
|
||||
req = urllib.request.Request(BASE + path, headers=SH)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
return json.loads(r.read().decode())
|
||||
|
||||
|
||||
def norm(s: str) -> str:
|
||||
s = s.lower()
|
||||
for a, b in (
|
||||
("á", "a"),
|
||||
("é", "e"),
|
||||
("í", "i"),
|
||||
("ó", "o"),
|
||||
("ú", "u"),
|
||||
("ñ", "n"),
|
||||
):
|
||||
s = s.replace(a, b)
|
||||
return re.sub(r"[^a-z0-9]+", " ", s).strip()
|
||||
|
||||
|
||||
def score_name(tenant_name: str, tenant_slug: str, account_name: str) -> int:
|
||||
t = norm(tenant_name)
|
||||
s = norm(tenant_slug.replace("-", " "))
|
||||
a = norm(account_name)
|
||||
if not a:
|
||||
return 0
|
||||
if a == t or a == s:
|
||||
return 100
|
||||
if t and (t in a or a in t):
|
||||
return 85
|
||||
if s and (s in a or a in s):
|
||||
return 80
|
||||
tt = {x for x in t.split() if len(x) > 2}
|
||||
aa = {x for x in a.split() if len(x) > 2}
|
||||
if not tt or not aa:
|
||||
return 0
|
||||
hit = len(tt & aa) / len(tt)
|
||||
if hit >= 0.8:
|
||||
return 70
|
||||
if hit >= 0.5:
|
||||
return 55
|
||||
return int(hit * 50)
|
||||
|
||||
|
||||
def top_candidates(tenants, account_name, n=3):
|
||||
scored = []
|
||||
for t in tenants:
|
||||
sc = score_name(t["name"], t["slug"], account_name)
|
||||
if sc > 0:
|
||||
scored.append((sc, t["name"]))
|
||||
scored.sort(key=lambda x: -x[0])
|
||||
return scored[:n]
|
||||
|
||||
|
||||
def meta_paginate(token: str, path: str):
|
||||
out = []
|
||||
url = f"{META}/{path}"
|
||||
sep = "&" if "?" in path else "?"
|
||||
url = url + sep + "access_token=" + urllib.parse.quote(token)
|
||||
while url:
|
||||
req = urllib.request.Request(url)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
j = json.loads(r.read().decode())
|
||||
out.extend(j.get("data") or [])
|
||||
url = (j.get("paging") or {}).get("next")
|
||||
return out
|
||||
|
||||
|
||||
def google_refresh(refresh: str) -> str:
|
||||
body = urllib.parse.urlencode(
|
||||
{
|
||||
"refresh_token": refresh,
|
||||
"client_id": ENV["GOOGLE_OAUTH_CLIENT_ID"],
|
||||
"client_secret": ENV["GOOGLE_OAUTH_CLIENT_SECRET"],
|
||||
"grant_type": "refresh_token",
|
||||
}
|
||||
).encode()
|
||||
req = urllib.request.Request(
|
||||
"https://oauth2.googleapis.com/token",
|
||||
data=body,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
return json.loads(r.read().decode())["access_token"]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
tenants = sb("/rest/v1/tenants?select=id,name,slug&order=name.asc")
|
||||
integs = sb(
|
||||
"/rest/v1/integrations?provider=in.(ga4,search_console,google_drive,google_ads,meta_ads,facebook,instagram)"
|
||||
"&select=tenant_id,provider,status,external_account_id,meta,error_message"
|
||||
)
|
||||
by = defaultdict(dict)
|
||||
for i in integs:
|
||||
by[i["tenant_id"]][i["provider"]] = i
|
||||
|
||||
grow = sb(
|
||||
"/rest/v1/agency_oauth?provider=eq.google_drive&select=access_token,refresh_token,account_email"
|
||||
)[0]
|
||||
mrow = sb(
|
||||
"/rest/v1/agency_oauth?provider=eq.meta_ads&select=access_token,account_email,scopes"
|
||||
)[0]
|
||||
gtoken = google_refresh(grow["refresh_token"])
|
||||
mtoken = mrow["access_token"]
|
||||
|
||||
print("=== TOKENS ===")
|
||||
print("Google:", grow.get("account_email"))
|
||||
print("Meta:", mrow.get("account_email"))
|
||||
|
||||
# Meta inventory
|
||||
ads = []
|
||||
for row in meta_paginate(
|
||||
mtoken, "me/adaccounts?fields=id,account_id,name&limit=200"
|
||||
):
|
||||
aid = row.get("account_id") or row["id"].replace("act_", "")
|
||||
ads.append({"id": f"act_{aid}", "name": row.get("name") or aid})
|
||||
pages = []
|
||||
igs = []
|
||||
for page in meta_paginate(
|
||||
mtoken,
|
||||
"me/accounts?fields=id,name,instagram_business_account{id,username,name}&limit=100",
|
||||
):
|
||||
pages.append({"id": page["id"], "name": page.get("name") or page["id"]})
|
||||
ig = page.get("instagram_business_account") or {}
|
||||
if ig.get("id"):
|
||||
igs.append(
|
||||
{
|
||||
"id": ig["id"],
|
||||
"name": f"@{ig['username']}"
|
||||
if ig.get("username")
|
||||
else (ig.get("name") or page.get("name") or ig["id"]),
|
||||
"page": page.get("name"),
|
||||
}
|
||||
)
|
||||
|
||||
print(f"\n=== META VISIBLE === ads={len(ads)} pages={len(pages)} ig={len(igs)}")
|
||||
print("Ad accounts:")
|
||||
for a in ads:
|
||||
cands = top_candidates(tenants, a["name"])
|
||||
linked = next(
|
||||
(
|
||||
t["name"]
|
||||
for t in tenants
|
||||
if (by[t["id"]].get("meta_ads") or {}).get("external_account_id")
|
||||
== a["id"]
|
||||
),
|
||||
None,
|
||||
)
|
||||
print(
|
||||
f" {a['name']!r} ({a['id']}) → linked={linked or 'NO'} best={cands}"
|
||||
)
|
||||
print("Pages:")
|
||||
for a in pages:
|
||||
cands = top_candidates(tenants, a["name"])
|
||||
linked = next(
|
||||
(
|
||||
t["name"]
|
||||
for t in tenants
|
||||
if (by[t["id"]].get("facebook") or {}).get("external_account_id")
|
||||
== a["id"]
|
||||
),
|
||||
None,
|
||||
)
|
||||
print(
|
||||
f" {a['name']!r} ({a['id']}) → linked={linked or 'NO'} best={cands}"
|
||||
)
|
||||
print("Instagram:")
|
||||
for a in igs:
|
||||
cands = top_candidates(tenants, a["name"]) + top_candidates(
|
||||
tenants, a.get("page") or ""
|
||||
)
|
||||
# dedupe keep best
|
||||
seen = {}
|
||||
for sc, name in cands:
|
||||
seen[name] = max(sc, seen.get(name, 0))
|
||||
cands = sorted(((v, k) for k, v in seen.items()), reverse=True)[:3]
|
||||
linked = next(
|
||||
(
|
||||
t["name"]
|
||||
for t in tenants
|
||||
if (by[t["id"]].get("instagram") or {}).get("external_account_id")
|
||||
== a["id"]
|
||||
),
|
||||
None,
|
||||
)
|
||||
print(
|
||||
f" {a['name']!r} page={a.get('page')!r} ({a['id']}) → linked={linked or 'NO'} best={cands}"
|
||||
)
|
||||
|
||||
# GA4 / GSC inventory
|
||||
req = urllib.request.Request(
|
||||
"https://analyticsadmin.googleapis.com/v1beta/accountSummaries",
|
||||
headers={"Authorization": f"Bearer {gtoken}"},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
summaries = json.loads(r.read().decode())
|
||||
ga4 = []
|
||||
for acc in summaries.get("accountSummaries") or []:
|
||||
for p in acc.get("propertySummaries") or []:
|
||||
if p.get("property"):
|
||||
ga4.append(
|
||||
{
|
||||
"id": p["property"],
|
||||
"name": p.get("displayName") or p["property"],
|
||||
"account": acc.get("displayName"),
|
||||
}
|
||||
)
|
||||
|
||||
req = urllib.request.Request(
|
||||
"https://www.googleapis.com/webmasters/v3/sites",
|
||||
headers={"Authorization": f"Bearer {gtoken}"},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
sites_j = json.loads(r.read().decode())
|
||||
gsc = [
|
||||
{"id": s["siteUrl"], "name": s["siteUrl"]}
|
||||
for s in (sites_j.get("siteEntry") or [])
|
||||
if s.get("siteUrl")
|
||||
]
|
||||
|
||||
print(f"\n=== GOOGLE VISIBLE === ga4={len(ga4)} gsc={len(gsc)}")
|
||||
|
||||
# Test Data API on first property
|
||||
if ga4:
|
||||
prop = ga4[0]["id"].replace("properties/", "")
|
||||
end = (datetime.now(timezone.utc) - timedelta(days=1)).date()
|
||||
start = end - timedelta(days=2)
|
||||
body = json.dumps(
|
||||
{
|
||||
"dateRanges": [{"startDate": str(start), "endDate": str(end)}],
|
||||
"dimensions": [{"name": "date"}],
|
||||
"metrics": [{"name": "sessions"}],
|
||||
}
|
||||
).encode()
|
||||
req = urllib.request.Request(
|
||||
f"https://analyticsdata.googleapis.com/v1beta/properties/{prop}:runReport",
|
||||
data=body,
|
||||
headers={
|
||||
"Authorization": f"Bearer {gtoken}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
print("Data API test: OK on", ga4[0]["name"], prop)
|
||||
except urllib.error.HTTPError as e:
|
||||
print("Data API test: FAIL", e.read().decode()[:250])
|
||||
|
||||
# Unmatched GA4 with best candidates (only those not linked)
|
||||
linked_ga4_ids = {
|
||||
(by[t["id"]].get("ga4") or {}).get("external_account_id")
|
||||
for t in tenants
|
||||
}
|
||||
linked_ga4_props = set()
|
||||
for t in tenants:
|
||||
meta = (by[t["id"]].get("ga4") or {}).get("meta") or {}
|
||||
if meta.get("property_id"):
|
||||
linked_ga4_props.add(meta["property_id"])
|
||||
|
||||
print("\n=== GA4 sin ficha clara (score<50 o no enlazado) — top candidatos ===")
|
||||
for a in ga4:
|
||||
linked = next(
|
||||
(
|
||||
t["name"]
|
||||
for t in tenants
|
||||
if (by[t["id"]].get("ga4") or {}).get("meta", {}).get("property_id")
|
||||
== a["id"]
|
||||
or (by[t["id"]].get("ga4") or {}).get("external_account_id")
|
||||
in (a["id"], a["name"])
|
||||
),
|
||||
None,
|
||||
)
|
||||
cands = top_candidates(tenants, a["name"])
|
||||
best = cands[0][0] if cands else 0
|
||||
if linked and best >= 50:
|
||||
continue
|
||||
if linked:
|
||||
continue
|
||||
print(f" GA4 {a['name']!r} ({a['id']}) acct={a.get('account')!r} best={cands}")
|
||||
|
||||
print("\n=== GSC sin ficha clara ===")
|
||||
for a in gsc:
|
||||
linked = next(
|
||||
(
|
||||
t["name"]
|
||||
for t in tenants
|
||||
if (by[t["id"]].get("search_console") or {}).get("external_account_id")
|
||||
== a["id"]
|
||||
),
|
||||
None,
|
||||
)
|
||||
if linked:
|
||||
continue
|
||||
cands = top_candidates(tenants, a["name"])
|
||||
# also try domain without sc-domain
|
||||
bare = a["name"].replace("sc-domain:", "").replace("https://", "").replace("http://", "").strip("/")
|
||||
cands2 = top_candidates(tenants, bare)
|
||||
merged = {}
|
||||
for sc, name in cands + cands2:
|
||||
merged[name] = max(sc, merged.get(name, 0))
|
||||
cands = sorted(((v, k) for k, v in merged.items()), reverse=True)[:3]
|
||||
print(f" GSC {a['name']!r} best={cands}")
|
||||
|
||||
# Per-tenant gaps with why
|
||||
print("\n=== POR CLIENTE: qué falta y por qué ===")
|
||||
providers = [
|
||||
"google_ads",
|
||||
"meta_ads",
|
||||
"ga4",
|
||||
"search_console",
|
||||
"google_drive",
|
||||
"facebook",
|
||||
"instagram",
|
||||
]
|
||||
|
||||
def ok(i):
|
||||
return bool(i and i.get("status") == "connected" and i.get("external_account_id"))
|
||||
|
||||
for t in tenants:
|
||||
m = by[t["id"]]
|
||||
gaps = []
|
||||
for p in providers:
|
||||
if ok(m.get(p)):
|
||||
continue
|
||||
if p == "meta_ads":
|
||||
cands = [(score_name(t["name"], t["slug"], a["name"]), a["name"], a["id"]) for a in ads]
|
||||
cands = [c for c in sorted(cands, reverse=True) if c[0] > 0][:2]
|
||||
if not ads:
|
||||
gaps.append("meta_ads: token no ve ad accounts")
|
||||
elif not cands:
|
||||
gaps.append("meta_ads: no hay ad account cuyo nombre se parezca (¿no está en las 9 visibles?)")
|
||||
else:
|
||||
gaps.append(f"meta_ads: no auto-match; candidatos {cands}")
|
||||
elif p == "facebook":
|
||||
cands = [(score_name(t["name"], t["slug"], a["name"]), a["name"], a["id"]) for a in pages]
|
||||
cands = [c for c in sorted(cands, reverse=True) if c[0] > 0][:2]
|
||||
if not cands:
|
||||
gaps.append("facebook: ninguna Page visible se parece al nombre")
|
||||
else:
|
||||
gaps.append(f"facebook: no auto-match; candidatos {cands}")
|
||||
elif p == "instagram":
|
||||
cands = []
|
||||
for a in igs:
|
||||
sc = max(
|
||||
score_name(t["name"], t["slug"], a["name"]),
|
||||
score_name(t["name"], t["slug"], a.get("page") or ""),
|
||||
)
|
||||
if sc > 0:
|
||||
cands.append((sc, a["name"], a.get("page"), a["id"]))
|
||||
cands = sorted(cands, reverse=True)[:2]
|
||||
if not cands:
|
||||
gaps.append("instagram: ningún @ / page visible se parece")
|
||||
else:
|
||||
gaps.append(f"instagram: no auto-match; candidatos {cands}")
|
||||
elif p == "ga4":
|
||||
cands = [(score_name(t["name"], t["slug"], a["name"]), a["name"], a["id"]) for a in ga4]
|
||||
cands = [c for c in sorted(cands, reverse=True) if c[0] > 0][:2]
|
||||
if not cands:
|
||||
gaps.append("ga4: ninguna propiedad visible se parece al nombre")
|
||||
else:
|
||||
gaps.append(f"ga4: no auto-match (umbral 50); candidatos {cands}")
|
||||
elif p == "search_console":
|
||||
cands = []
|
||||
for a in gsc:
|
||||
bare = a["name"].replace("sc-domain:", "").replace("https://", "").replace("http://", "").strip("/")
|
||||
sc = max(
|
||||
score_name(t["name"], t["slug"], a["name"]),
|
||||
score_name(t["name"], t["slug"], bare),
|
||||
)
|
||||
if sc > 0:
|
||||
cands.append((sc, a["name"]))
|
||||
cands = sorted(cands, reverse=True)[:2]
|
||||
if not cands:
|
||||
gaps.append("gsc: ningún site visible se parece")
|
||||
else:
|
||||
gaps.append(f"gsc: no auto-match (umbral 40); candidatos {cands}")
|
||||
elif p == "google_drive":
|
||||
gaps.append("drive: sin carpeta auto-matched (revisar nombre carpeta Drive)")
|
||||
elif p == "google_ads":
|
||||
gaps.append("google_ads: sin cuenta enlazada (MCC / nombre)")
|
||||
if gaps:
|
||||
print(f"\n{t['name']}:")
|
||||
for g in gaps:
|
||||
print(f" - {g}")
|
||||
else:
|
||||
print(f"\n{t['name']}: COMPLETO (todos los canales con ID)")
|
||||
|
||||
# BPS GSC permission probe
|
||||
bps = next(t for t in tenants if t["name"] == "BPS")
|
||||
gsc_id = (by[bps["id"]].get("search_console") or {}).get("external_account_id")
|
||||
if gsc_id:
|
||||
end = (datetime.now(timezone.utc) - timedelta(days=1)).date()
|
||||
start = end - timedelta(days=3)
|
||||
body = json.dumps(
|
||||
{"startDate": str(start), "endDate": str(end), "dimensions": ["date"]}
|
||||
).encode()
|
||||
req = urllib.request.Request(
|
||||
f"https://www.googleapis.com/webmasters/v3/sites/{urllib.parse.quote(gsc_id, safe='')}/searchAnalytics/query",
|
||||
data=body,
|
||||
headers={
|
||||
"Authorization": f"Bearer {gtoken}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
print("\nBPS GSC probe: OK")
|
||||
except urllib.error.HTTPError as e:
|
||||
print("\nBPS GSC probe: FAIL", e.read().decode()[:220])
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user