A cosa serve una memoria perfetta, se la interroghi sempre troppo tardi?
Tre script che rendono un second brain Obsidian predittivo: radar scadenze a 30 giorni, auto-riparazione link, forecast KPI. Architettura e codice scaricabile (paid).
Come ho fatto passare il mio vault da archivio del passato a sistema che anticipa scadenze, link rotti e numeri fermi. I tre script dentro il deep dive.
Venerdì sera, le 23 passate, sto fissando il file di stato del mio vault con il caffè ormai freddo accanto. C’è una cosa importante ferma lì da diciannove giorni. Una decisione presa con un cliente a inizio mese, che lo stato del progetto non aveva mai recepito. Diciannove giorni in cui ho riletto quel file decine di volte senza vederla, perché era scritta e basta, ferma, muta. Il mio gatto è salito sulla scrivania, si è seduto sulla tastiera come fa sempre quando lavoro troppo, e io continuavo a pensare la stessa cosa: ho un second brain con tre milioni di parole, e mi sono perso una riga che contava.
Il problema non era la memoria. La memoria era perfetta. Il problema è che un archivio aspetta che tu gli faccia la domanda giusta al momento giusto. E al momento giusto, di solito, sei distratto.
In sintesi: un second brain predittivo è un sistema di note dove l’AI non si limita a registrare cosa è successo, ma legge i tuoi file e ti avvisa di scadenze, link rotti e numeri fermi prima che diventino un problema. Si costruisce con tre script che girano sul tuo vault: un radar delle scadenze a 30 giorni, un riparatore di link, un proiettore di KPI. Nessun rito manuale in più.
Cosa è un second brain predittivo, e perché il tuo oggi non lo è
Quasi tutti i second brain, il mio compreso fino a tre settimane fa, sono archivisti bravissimi e analisti pessimi. Catturano tutto: ogni nota, ogni decisione, ogni numero. Li ritrovi quando li cerchi. Ma cercarli è compito tuo, e il tuo cervello è la parte fragile del sistema.
Funziona così. Scrivi nel vault che un preventivo scade tra trenta giorni. Lo scrivi bene, in chiaro, nel posto giusto. Poi passano i giorni, tu lavori su altro, e quella scadenza resta sepolta tra cinquemila file finché o ti viene in mente, o è tardi. La nota c’era. Nessuno l’ha letta al momento giusto. Lo stesso vale per un link interno che si rompe quando sposti un file, per un numero di crescita che hai aggiornato l’ultima volta a maggio e che ormai mente, per un lead da richiamare che era scritto in una riga che non guardi da settimane.
La memoria perfetta non ti salva da niente, se la interroghi solo quando ti ricordi di interrogarla. Il salto non è ricordare di più. È smettere di aspettare la tua domanda.
Il principio, ogni sorpresa merita un check che la misura
Quando ho ragionato su come chiudere il buco, la tentazione facile era aggiungere riti. Il lunedì controllo le scadenze, il mercoledì verifico i link, la domenica aggiorno i numeri. L’ho già provato altre volte, con la mia agenda e con le persone del team. I riti settimanali falliscono in silenzio. Salti una settimana perché sei in trasferta, ne salti due perché eri sotto consegna, e il sistema non protesta. Tace, esattamente come taceva lo stato fermo da diciannove giorni.
Così ho cambiato strada. Niente riti. Un principio solo: ogni categoria di sorpresa ha diritto a un controllo automatico che la misura e fa rumore quando qualcosa non va. Una scadenza che si avvicina è una sorpresa. Un link che si rompe è una sorpresa. Un numero che invecchia è una sorpresa. Tre tipi di sorpresa, tre controlli che girano da soli sul vault e mi mettono davanti agli occhi quello che sta per diventare un problema. La differenza tra un rito e un check è che il rito chiede disciplina a me, il check no. Il check misura l’età delle cose e me le sbatte in faccia.
Tre script, scritti in un paio di sere tra l’undici e il dodici giugno, hanno fatto questo lavoro. Te li spiego uno per uno, con la logica dentro. Il codice completo e i file da copiare li trovi nel deep dive in fondo, e un prompt per costruire il primo sul tuo vault è in coda a questo post, gratis per pochi giorni.
Come funzionano i tre sistemi, uno alla volta
Il radar che guarda trenta giorni avanti
Il primo si chiama horizon, ed è il cuore. È uno script che ogni volta che gira legge tre posti del vault e produce un solo file, horizon.md, con gli eventi dei prossimi trenta giorni divisi in fasce: rosso entro tre giorni, arancio entro due settimane, giallo entro un mese.
I tre posti da cui pesca sono diversi apposta. Il primo sono i file di stato e la lista dei task del vault, che si aggiornano da soli a fine sessione: lo script ne estrae le date, anche quelle buttate lì di fretta. Il secondo sono le scadenze contrattuali e strategiche, e quelle non le scrivo a mano in un file: vivono nel gestionale dell’agenzia, Agency OS, dove tengo clienti, progetti, preventivi e le loro date. Via MCP il vault le legge direttamente da lì e le deposita nel registro che il radar consulta, senza che io ricopi niente. Come ho collegato Agency OS dentro Cowork con un MCP l’ho raccontato in Agency OS dentro Cowork. Il terzo è la lista delle persone con cui ho un rapporto aperto, dove tengo una colonna “prossimo contatto”: da lì escono i follow-up, da soli.
La parte su cui ho lavorato di più è il filtro anti-abbaglio. Un vault è pieno di numeri che sembrano date e non lo sono: “round 1/2”, “9/11 video da montare”, “fase 3”. Se prendi ogni 1/2 per il primo febbraio, il radar diventa rumore e smetti di guardarlo. Così lo script tiene due liste nere: parole che, se vengono prima di un numero-data, lo squalificano (round, fase, sessione, video), e parole che lo squalificano se vengono dopo (video, post, task, lead). E pesa il contesto: se nella riga c’è “entro”, “scadenza”, “call”, “follow-up”, alza la confidenza; se non c’è, segna la data con un punto interrogativo, così so che è un sospetto, non una certezza. Al primo giro vero il radar ha tirato su ventotto fonti reali senza un solo falso positivo rimasto.
L’effetto, la mattina, è che apro un file e vedo i prossimi trenta giorni in ordine, con dentro le scadenze dei contratti, i follow-up in ritardo e le milestone, ciascuna con scritto da quale file arriva. Non devo più ricordarmi di chiedere. Me lo trovo davanti.
Sul perché tengo separate le scadenze del gestionale dalle date pescate in automatico dai file di stato ho scritto la logica nel post di mercoledì scorso, i sei file che danno memoria a Claude. Quello spiega l’architettura statica della memoria, questo è il piano sopra: cosa ci faccio leggere e quando.
Il vault che si ripara da solo
Il secondo sistema è meno appariscente e più liberatorio. Ogni volta che sposti o rinomini un file in un sistema di note collegate, qualche link interno si rompe. Da solo è una banalità, su cinquemila file diventa una palude. E i link rotti sono il modo più stupido in cui un second brain perde valore: l’informazione c’è ancora, solo che il sentiero per arrivarci è interrotto.
Lo script trova i link markdown che puntano a file inesistenti e prova a ripararli con strategie ordinate dalla più sicura. La prima: se nel vault esiste un solo file con quel nome da qualche altra parte, vuol dire che l’ho semplicemente spostato, e lo script riscrive il percorso giusto. Copre da sola il caso più frequente, il post che migra dalla cartella bozze a quella dei pubblicati. La seconda gestisce i link a un indice che ha cambiato nome. Ci sono anche due strategie più invasive, che cancellano righe o riscrivono link a sistemi dismessi, ma quelle le tengo a parte.
La scelta che rende il sistema usabile senza paura è una sola opzione: la modalità sicura. In quella modalità lo script applica solo le due riparazioni che spostano percorsi e non cancellano mai una riga. È la modalità che gira in automatico nei miei cicli di lavoro. Le riparazioni distruttive non vengono eseguite di nascosto: finiscono in una lista “da guardare a mano”, etichettate, così le decido io con calma. Il primo giro reale in modalità sicura ha sistemato nove link, comprese le pubblicazioni di quella stessa mattina, senza che io toccassi niente. La regola di fondo è che un sistema che si auto-corregge ha il permesso di spostare, mai quello di buttare via in silenzio.
I numeri che si proiettano da soli
Il terzo trasforma i numeri da fotografia a film. Ogni settimana annoto gli iscritti e i follower dei canali in un file, in fondo, senza mai cancellare i vecchi. Da solo è un mucchio di fotografie datate. Lo script le rilegge tutte e le impila in serie storiche, una per canale, e poi su ognuna calcola una proiezione.
La proiezione è volutamente stupida, nel senso buono: niente modelli oscuri. Prende il ritmo degli ultimi punti (quanta crescita al giorno), lo ancora all’ultimo dato vero (mai a oggi, perché oggi non ho ancora misurato niente), e stima quando toccherò la prossima soglia. Se il canale è fermo o cala, non inventa una data: scrive che il trend è piatto. E ogni stima è marcata come indicativa, non come promessa, perché il giorno in cui inizi a credere alle tue proiezioni come fossero certe hai smesso di fare marketing e hai iniziato a leggere l’oroscopo. Concretamente, l’ultima volta mi ha detto duecentocinquanta iscritti Substack intorno al quindici giugno e quattrocento follower su un altro canale verso il ventiquattro. Numeri che mi servono per decidere dove spingere, non per rassicurarmi.
Tre script, tre tipi di sorpresa coperti. Sotto questi tre, nel sistema completo, ce ne sono altri due più piccoli che lavorano in appoggio: la lista delle persone che alimenta il radar dei follow-up, e un piccolo meccanismo che mi propone una cosa nuova da ricordare alla fine di ogni sessione. Ma i tre portanti sono questi, e da soli cambiano il mestiere del vault.
Se non sei ancora iscritto, qui sotto puoi farlo. Scrivo ogni mercoledì un sistema così, con dentro il codice e i file da copiare. Casi reali, niente teoria.
Il caso concreto, il follow-up che non si è perso
Faccio l’esempio più banale e più costoso, perché è quello che capita a tutti. Un lead arrivato a fine maggio, una conversazione buona, e l’impegno mio di richiamare “tra una decina di giorni”. Scritto dove? In una riga, dentro il file di una persona, sepolto. Nel mondo archivista quel richiamo dipende solo dalla mia memoria, e la mia memoria a dieci giorni di distanza, sotto tre consegne, vale poco.
Con il radar quella riga ha una colonna “prossimo contatto” con una data. Lo script la legge, e il giorno in cui il richiamo entra nei tre giorni quel follow-up compare in cima a horizon.md, in rosso, senza che io abbia chiesto niente. Prima il sistema teneva traccia. Adesso il sistema avvisa. La differenza, su un lead, è la differenza tra una nota ordinata e un contratto firmato.
Il prima e dopo che misuro non è sui lead, perché lì conta il caso singolo, è sullo stato del vault. Prima: un file di stato fermo diciannove giorni senza che nessuno se ne accorgesse. Dopo, con il controllo che misura l’età dei file e protesta sopra una soglia: zero file fermi, perché quando uno invecchia troppo lo vedo subito.
I quattro errori che ho fatto, così te li risparmi
Il primo: dare in pasto all’AI tutto il futuro. La tentazione, dopo i tre milioni di parole, è scaricare ogni scadenza possibile dentro un unico file e dire “adesso sai tutto”. Il radar annega esattamente come annega la memoria quando le versi addosso troppo. Le scadenze contano nel radar perché sono poche e curate nel gestionale, non perché le ho scaricate tutte.
Il secondo: lasciare che uno script si auto-corregga cancellando. La prima versione del riparatore poteva rimuovere righe in autonomia. Per due giorni ho avuto paura ad accenderlo. La modalità sicura, che sposta e non butta mai, è ciò che mi ha fatto fidare abbastanza da lasciarlo girare ogni giorno. Un sistema autonomo si guadagna la fiducia rinunciando al potere di distruggere.
Il terzo: confondere la proiezione con la promessa. La prima volta che ho letto “duecentocinquanta iscritti il quindici giugno” mi sono quasi rilassato. Sbagliato. La proiezione ti dice dove stai andando al ritmo attuale, e serve proprio a cambiarlo se non ti piace, non a sederti.
Il quarto, il più subdolo: rispondere al problema con un nuovo rito. Ogni volta che ti accorgi di un buco, l’istinto è promettere a te stesso un controllo manuale settimanale. Non reggerà. La domanda giusta non è “quando lo controllo”, è “quale check può misurarlo al posto mio”.
Sotto trovi il deep dive completo: i tre script veri e propri pronti da copiare e adattare al tuo vault, lo scheletro del registro delle scadenze, le regole esatte del filtro anti-abbaglio del radar, e i numeri grezzi dei primi giri. Per leggerlo serve essere maikers paganti (cinque euro al mese o cinquanta l’anno) o AI mAIker.
Qui sotto c’è tutto quello che ti serve per non riscrivere niente da zero. Sono i file reali che girano sul mio vault, ripuliti dai riferimenti personali.
Lo script del radar (horizon)
Struttura completa: le tre funzioni di lettura (file di stato, registro, persone), il parser delle date con le quattro forme riconosciute (data piena, intervallo giorno-giorno, giorno/mese con scivolamento all’anno dopo se la data è già passata, e numero di settimana solo sulle righe-task aperte), le due liste nere parola-prima e parola-dopo per intero, e la funzione che de-duplica gli eventi quando la stessa scadenza compare in due fonti, tenendo la versione curata a mano sopra quella pescata in automatico. In coda lo script stampa una riga riassuntiva che un secondo controllo di salute del vault legge per sapere quanti eventi rossi ci sono.
#!/usr/bin/env python3
"""horizon-scan.py — Radar predittivo: eventi futuri 0-30gg dal vault.
Tre sorgenti (priorità de-dup: registro > radar > scan):
1. SCAN — date parsate da STATE.md/TASKS.md attivi (stesso perimetro del
check governance-freshness: root + aree + clienti FULL)
2. REGISTRO — 3_Risorse/sistema/horizon-eventi.md (eventi contrattuali e
strategici curati, data ISO, prefisso ~ = stima)
3. RADAR — colonna "Prossimo touch" del Radar relazionale in
3_Risorse/sistema/people-index.md (sezione assente → 0 eventi)
Output: 3_Risorse/sistema/horizon.md (fasce 🔴 ≤3gg · 🟠 4-14 · 🟡 15-30 ·
⚫ scaduti registro/radar · 📜 oltre orizzonte) — MAI editare a mano.
Ultima riga stdout machine-parsable per vault-health-check (sezione 10):
HORIZON_SUMMARY red=N orange=N yellow=N overdue=N total=N
Parser anti-falsi-positivi: finestra 0-30gg + blacklist parola-prima/dopo
(frazioni "round 1/2", rapporti "9/11 video") + esclusione righe done/fence.
W##: solo da righe-task, solo settimane future, data = lunedì ISO week
(stesso anno ISO corrente — niente rollover dicembre→gennaio, caso accettato).
Flag: --dry-run (stampa senza scrivere horizon.md).
Regole: 3_Risorse/conventions.md §"Horizon". Idempotente intra-day.
"""
import re
import sys
from datetime import date, timedelta
from pathlib import Path
VAULT = Path('<IL_TUO_VAULT>') # <-- cambia con il path della tua cartella di note
REGISTRY = VAULT / '3_Risorse' / 'sistema' / 'horizon-eventi.md'
PEOPLE = VAULT / '3_Risorse' / 'sistema' / 'people-index.md'
OUTPUT = VAULT / '3_Risorse' / 'sistema' / 'horizon.md'
WINDOW_DAYS = 30
DRY_RUN = '--dry-run' in sys.argv
SKIP_DIRS = {'.git', '.obsidian', 'node_modules', '__pycache__', '4_Archivio'}
RE_ISO = re.compile(r'\b(\d{4})-(\d{2})-(\d{2})\b')
RE_RANGE = re.compile(r'(?<![\d/])(\d{1,2})-(\d{1,2})/(\d{1,2})(?![\d/])')
RE_GGM = re.compile(r'(?<![\d/-])(\d{1,2})/(\d{1,2})(?:/(\d{4}))?(?![\d/])')
RE_WEEK = re.compile(r'\bW(\d{2})\b')
RE_WEEK_RANGE = re.compile(r'\bW\d{2}-W\d{2}\b')
RE_TIME = re.compile(r'\b(\d{1,2}):(\d{2})\b')
DAYNAMES = {'lun': 0, 'mar': 1, 'mer': 2, 'gio': 3, 'ven': 4, 'sab': 5, 'dom': 6}
DAYNAMES_IT = ['lun', 'mar', 'mer', 'gio', 'ven', 'sab', 'dom']
BLACK_BEFORE = {
'round', 'sess', 'sessione', 'sessioni', 'fase', 'step', 'parte',
'episodio', 'ep', 'video', 'giorno', 'modulo', 'lezione', 'puntata',
'batch', 'lotto', 'draft', 'rev', 'v', 'vol', 'q', 'pag', 'pagina',
'cap', 'capitolo', 'slide', 'punto', 'task', 'sprint', 'settimana',
'opzione', 'scenario', 'score', 'voto', 'tracker', 'rapporto',
}
BLACK_AFTER = {
'video', 'post', 'task', 'azioni', 'punti', 'clienti', 'lead',
'preventivi', 'sessioni', 'file', 'item', 'contenuti', 'candidati',
'persone', 'slot', 'abbonati', 'mesi', 'paid', 'sveglie', 'sveglia',
'outreach', 'giorni', 'volte',
}
CONTEXT_KEYWORDS = {
'entro', 'scadenza', 'deadline', 'call', 'meeting', 'meet', 'sess',
'sessione', 'reminder', 'promemoria', 'follow-up', 'invio', 'consegna',
'lancio', 'rinnovo', 'paywall', 'review', 'quarterly', 'erogazione',
'kickoff', 'presentazione', 'shooting', 'evento', 'conferma', 'fattura',
'pagamento', 'webinar', 'workshop', 'meetup', 'triage', 'decisione',
'appuntamento', 'fissata', 'fissato', 'due',
}
STOPWORDS_OVERLAP = {'entro', 'sessione', 'sess', 'call', 'invio', 'follow',
'scadenza', 'della', 'dello', 'review', 'academy'}
def governance_targets() -> list:
"""STATE/TASKS attivi: root + aree + clienti FULL (perimetro governance-freshness)."""
targets = []
for name in ('STATE.md', 'TASKS.md'):
p = VAULT / name
if p.exists():
targets.append((p, 'root'))
for f in VAULT.glob(f'2_Aree/*/{name}'):
targets.append((f, f.parts[len(VAULT.parts) + 1]))
for f in VAULT.glob(f'2_Aree/*/1_Progetti/_clienti/*/{name}'):
area = f.parts[len(VAULT.parts) + 1]
cliente = f.parts[len(VAULT.parts) + 4]
targets.append((f, f'{area}/{cliente}'))
return [(f, scope) for f, scope in targets
if not any(s in f.parts for s in SKIP_DIRS)]
def clean_label(line: str) -> str:
"""Riga → label evento leggibile (max 90 char)."""
s = line.strip()
s = re.sub(r'^[-*]\s*\[[ x~]\]\s*', '', s) # checkbox
s = re.sub(r'^[-*>|]\s*', '', s) # bullet/quote/pipe
s = re.sub(r'\[([^\]]*)\]\([^)]+\)', r'\1', s) # link md → testo
s = s.replace('**', '').replace('`', '')
s = re.sub(r'\s+', ' ', s).strip(' ·|')
return (s[:90] + '…') if len(s) > 90 else s
def word_before(line: str, start: int) -> str:
m = re.search(r'([\w-]+)\W{0,3}$', line[:start])
return m.group(1).lower() if m else ''
def word_after(line: str, end: int) -> str:
m = re.match(r'\W{0,3}([\w-]+)', line[end:])
return m.group(1).lower() if m else ''
def find_time(line: str, end: int) -> str:
m = RE_TIME.search(line[end:end + 12])
return f'{int(m.group(1)):02d}:{m.group(2)}' if m else ''
def valid_date(y: int, mo: int, d: int):
try:
return date(y, mo, d)
except ValueError:
return None
def parse_dates_in_line(line: str, today: date) -> list:
"""Estrae candidate-date future (0-30gg) da una riga. Ritorna [{date, time, note, confidence}]."""
out = []
consumed = [] # span già matchati (range prima di GG/M)
low = line.lower()
has_keyword = any(k in low for k in CONTEXT_KEYWORDS)
def overlaps(a, b):
return any(not (b <= s or a >= e) for s, e in consumed)
# 1. ISO
for m in RE_ISO.finditer(line):
d = valid_date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
if d:
out.append({'date': d, 'span': m.span(), 'note': '',
'confidence': 'alta' if has_keyword else 'media'})
consumed.append(m.span())
# 2. Range GG-GG/M (prima di GG/M)
for m in RE_RANGE.finditer(line):
if overlaps(*m.span()):
continue
if word_before(line, m.start()) in BLACK_BEFORE or word_after(line, m.end()) in BLACK_AFTER:
continue
d1, d2, mo = int(m.group(1)), int(m.group(2)), int(m.group(3))
d = valid_date(today.year, mo, d1)
if d and d < today:
d = valid_date(today.year + 1, mo, d1)
if d:
out.append({'date': d, 'span': m.span(),
'note': f'(range {d1}-{d2}/{mo})',
'confidence': 'alta' if has_keyword else 'media'})
consumed.append(m.span())
# 3. GG/M
for m in RE_GGM.finditer(line):
if overlaps(*m.span()):
continue
gg, mo = int(m.group(1)), int(m.group(2))
yy = int(m.group(3)) if m.group(3) else today.year
if not (1 <= gg <= 31 and 1 <= mo <= 12):
continue
d = valid_date(yy, mo, gg)
if d and not m.group(3) and d < today:
d = valid_date(yy + 1, mo, gg)
if not d:
continue
wb = word_before(line, m.start())
wa = word_after(line, m.end())
dayname = wb if wb in DAYNAMES else ''
if not dayname:
if wb in BLACK_BEFORE or wa in BLACK_AFTER:
continue
note = ''
conf = 'alta' if (dayname or has_keyword) else 'media'
if dayname and d.weekday() != DAYNAMES[dayname]:
note = '(⚠️ giorno dichiarato non coincide)'
out.append({'date': d, 'span': m.span(), 'note': note, 'confidence': conf})
consumed.append(m.span())
# 4. W## (solo righe-task aperte, settimane future, no range W-W)
if line.lstrip().startswith('- [ ]') and not RE_WEEK_RANGE.search(line):
cur_week = today.isocalendar()[1]
for m in RE_WEEK.finditer(line):
w = int(m.group(1))
if w <= cur_week or w > 53:
continue
try:
d = date.fromisocalendar(today.year, w, 1)
except ValueError:
continue
sun = d + timedelta(days=6)
out.append({'date': d, 'span': m.span(),
'note': f'(settimana W{w}, {d.day}/{d.month}-{sun.day}/{sun.month})',
'confidence': 'media'})
# finestra + orario
final = []
for c in out:
delta = (c['date'] - today).days
if 0 <= delta <= WINDOW_DAYS:
c['time'] = find_time(line, c['span'][1])
final.append(c)
return final
def scan_governance(today: date) -> list:
events = []
for f, scope in governance_targets():
try:
lines = f.read_text(encoding='utf-8').split('\n')
except (UnicodeDecodeError, OSError):
continue
in_front = False
in_fence = False
in_done = False
for i, line in enumerate(lines, 1):
if i == 1 and line.strip() == '---':
in_front = True
continue
if in_front:
if line.strip() == '---':
in_front = False
continue
if line.lstrip().startswith('```'):
in_fence = not in_fence
continue
if in_fence:
continue
if re.match(r'^#{1,4}\s', line):
in_done = bool(re.search(r'✅\s*Done|Done recenti|Archivio', line))
if in_done:
continue
ls = line.lstrip()
if ls.startswith(('- [x]', '* [x]', '>')) or 'log-archive' in line:
continue
for c in parse_dates_in_line(line, today):
events.append({
'date': c['date'], 'time': c['time'],
'label': clean_label(line),
'note': c['note'], 'confidence': c['confidence'],
'scope': scope, 'kind': 'scan',
'sources': [f'{f.relative_to(VAULT)}:{i}'],
})
return events
RE_REG_ROW = re.compile(r'^\|\s*(~?)(\d{4}-\d{2}-\d{2})\s*\|([^|]+)\|([^|]+)\|([^|]*)\|')
def load_registry(today: date) -> list:
"""Righe | [~]YYYY-MM-DD | evento | scope | note | della sezione ## Eventi."""
events = []
if not REGISTRY.exists():
return events
in_events = False
for i, line in enumerate(REGISTRY.read_text(encoding='utf-8').split('\n'), 1):
if line.startswith('## '):
in_events = line.strip() == '## Eventi'
continue
if not in_events:
continue
m = RE_REG_ROW.match(line.strip())
if not m:
continue
d = valid_date(*map(int, m.group(2).split('-')))
if not d:
continue
note = m.group(5).strip()
if m.group(1) == '~':
note = ('(stima) ' + note).strip()
events.append({
'date': d, 'time': '', 'label': m.group(3).strip(),
'note': note, 'confidence': 'alta',
'scope': m.group(4).strip(), 'kind': 'registro',
'sources': [f'{REGISTRY.relative_to(VAULT)}:{i}'],
})
return events
def load_radar(today: date) -> list:
"""Colonna 'Prossimo touch' (4ª) del Radar relazionale in people-index.md."""
events = []
if not PEOPLE.exists():
return events
in_radar = False
for i, line in enumerate(PEOPLE.read_text(encoding='utf-8').split('\n'), 1):
if line.startswith('## '):
in_radar = 'Radar relazionale' in line
continue
if not in_radar or not line.strip().startswith('|'):
continue
cells = [c.strip() for c in line.strip().strip('|').split('|')]
if len(cells) < 4 or cells[0] in ('Persona', '---') or set(cells[0]) <= {'-'}:
continue
m = re.match(r'^(~?)(\d{4})-(\d{2})-(\d{2})$', cells[3])
if not m:
continue
d = valid_date(int(m.group(2)), int(m.group(3)), int(m.group(4)))
if not d:
continue
events.append({
'date': d, 'time': '',
'label': f'Follow-up {cells[0]} ({cells[1]})',
'note': '(stima)' if m.group(1) == '~' else '',
'confidence': 'alta', 'scope': 'radar',
'kind': 'radar',
'sources': [f'{PEOPLE.relative_to(VAULT)}:{i}'],
})
return events
def token_overlap(a: str, b: str) -> bool:
t = lambda s: {w for w in re.findall(r'[a-zà-ù0-9]{4,}', s.lower())} - STOPWORDS_OVERLAP
return bool(t(a) & t(b))
def dedupe(events: list) -> list:
"""Priorità registro > radar > scan; merge stessa data + token overlap."""
order = {'registro': 0, 'radar': 1, 'scan': 2}
kept = []
for ev in sorted(events, key=lambda e: (order[e['kind']], e['date'])):
twin = next((k for k in kept
if k['date'] == ev['date'] and token_overlap(k['label'], ev['label'])), None)
if twin:
twin['sources'].extend(ev['sources'])
if not twin['time'] and ev['time']:
twin['time'] = ev['time']
else:
kept.append(ev)
return kept
def fmt_event(ev) -> str:
d = ev['date']
day = f"{DAYNAMES_IT[d.weekday()]} {d.day}/{d.month}"
t = f" {ev['time']}" if ev['time'] else ''
q = ' ·?' if ev['confidence'] == 'media' else ''
note = f" {ev['note']}" if ev['note'] else ''
srcs = ' + '.join(list(dict.fromkeys(ev['sources']))[:3])
return f"- **{day}{t}** — {ev['label']}{note} · 📍 {ev['scope']} · fonte: `{srcs}`{q}"
def render_horizon(events, overdue, beyond, today: date, n_targets: int) -> str:
end = today + timedelta(days=WINDOW_DAYS)
red = [e for e in events if (e['date'] - today).days <= 3]
orange = [e for e in events if 3 < (e['date'] - today).days <= 14]
yellow = [e for e in events if 14 < (e['date'] - today).days <= 30]
L = []
L.append('---')
L.append('tipo: report')
L.append('audience: human+llm')
L.append('generato-da: horizon-scan.py')
L.append(f'freshness: {today.isoformat()}')
L.append('tldr: "Eventi futuri 0-30gg da TASKS/STATE attivi + registro horizon-eventi + radar relazionale. Rigenerato dal health-check (sezione 10). NON editare a mano."')
L.append('---')
L.append('')
L.append('# ⏰ Horizon — prossimi 30 giorni')
L.append('')
L.append(f'**Generato**: {today.isoformat()} · finestra {today.isoformat()} → {end.isoformat()} · fonti: {n_targets} STATE/TASKS + registro + radar')
L.append('')
for title, items in (('## 🔴 Entro 3 giorni', red),
('## 🟠 4-14 giorni', orange),
('## 🟡 15-30 giorni', yellow)):
L.append(title)
if items:
L.extend(fmt_event(e) for e in sorted(items, key=lambda e: (e['date'], e['time'])))
else:
L.append('- (nessuno)')
L.append('')
L.append('## ⚫ Follow-up / eventi scaduti (registro+radar — da ri-datare o archiviare)')
if overdue:
L.extend(fmt_event(e) for e in sorted(overdue, key=lambda e: e['date']))
else:
L.append('- (nessuno)')
L.append('')
L.append('## 📜 Oltre l\'orizzonte (>30gg, solo registro)')
if beyond:
L.append(' · '.join(f"{e['date'].isoformat()} {('~ ' if '(stima)' in e['note'] else '')}{e['label'][:60]}" for e in sorted(beyond, key=lambda e: e['date'])))
else:
L.append('- (nessuno)')
L.append('')
L.append(f'> ·? = data trovata senza keyword di contesto (confidence media). Regole parser: `scripts/horizon-scan.py` docstring · conventions §"Horizon".')
L.append('')
return '\n'.join(L)
def main():
today = date.today()
targets = governance_targets()
scan = scan_governance(today)
registry_all = load_registry(today)
radar_all = load_radar(today)
# finestra per registro/radar: dentro 0-30 → fasce; passato → overdue; oltre → beyond (solo registro)
in_window = lambda e: 0 <= (e['date'] - today).days <= WINDOW_DAYS
registry = [e for e in registry_all if in_window(e)]
radar = [e for e in radar_all if in_window(e)]
overdue = [e for e in registry_all + radar_all if (e['date'] - today).days < 0]
beyond = [e for e in registry_all if (e['date'] - today).days > WINDOW_DAYS]
events = dedupe(registry + radar + scan)
out = render_horizon(events, overdue, beyond, today, len(targets))
if DRY_RUN:
print(out)
else:
OUTPUT.write_text(out, encoding='utf-8')
print(f'✅ scritto {OUTPUT.relative_to(VAULT)}')
red = sum(1 for e in events if (e['date'] - today).days <= 3)
orange = sum(1 for e in events if 3 < (e['date'] - today).days <= 14)
yellow = sum(1 for e in events if 14 < (e['date'] - today).days <= 30)
print(f'HORIZON_SUMMARY red={red} orange={orange} yellow={yellow} overdue={len(overdue)} total={len(events)}')
if __name__ == '__main__':
main()
Lo script che ripara i link
Le quattro strategie in ordine di sicurezza, la modalità sicura che applica solo le prime due, e il dettaglio tecnico che evita il bug classico: quando rimuove righe lo fa dal basso verso l’alto, così non sballa la numerazione delle righe successive.
#!/usr/bin/env python3
"""
fix-broken-links.py — Fix automatico dei link rotti in vault.
4 strategie (applicate in ordine):
A) SAFE_RENAME: link target con basename uguale esistente altrove nel vault
(1 solo match) → riscrivi path. Copre anche il caso ricorrente
"post spostato bozze/<slug>/ → pubblicati/<slug>/" (basename unico).
B) README_TO_INDEX: link a `README.md` dove nella stessa cartella esiste
`INDEX.md` → sostituisce nome file.
C) DECOMMISSIONED_SERVICE_SKELETON: link a servizi decommissionati
(google-ads, seo, social-organico) se il target non esiste →
**rimuove la riga intera** nel file sorgente (solo bullet/tabella).
D) PHANTOM_SKILL: link a skill di area con prefisso noto ma file inesistente
→ unwrap `[testo](path)` → `testo` in backtick.
Casi non risolvibili auto → riportati come MANUAL_REVIEW.
Flag:
--dry-run anteprima, non scrive nulla
--safe-only applica SOLO A+B (riscrittura path, mai rimozione righe né
unwrap). Per uso AUTOMATICO nei cicli (aggiorna-lavoro step 6,
regola in conventions §"Auto-fix sicuri"). I C/D restano in
manual_review etichettati "(auto-fixabile con run completo)".
Limite noto: copre solo target .md — i link ad assets (immagini ecc.) non
vengono scansionati.
Idempotente.
"""
import re
import sys
from pathlib import Path
from collections import defaultdict
VAULT = Path('<IL_TUO_VAULT>') # <-- cambia con il path della tua cartella di note
SKIP_DIRS = {'.git', '.obsidian', 'node_modules', '_Inbox',
'__pycache__', '.venv', 'venv', 'notion-imports',
'skills-backup', # backup prompt/skill con template pseudo-link
'reports'} # output rigenerati di script (timestamped)
DRY_RUN = '--dry-run' in sys.argv
SAFE_ONLY = '--safe-only' in sys.argv # solo strategie A+B (mai remove_line/unwrap)
MD_LINK_LINE = re.compile(r'\[([^\]]*)\]\(([^)]+)\)')
# Servizi scheletro noti decommissionati
DECOMMISSIONED_SERVICES = {
'google-ads', 'seo', 'social-organico',
# NOTE: meta-ads e email-marketing possono esistere in alcuni clienti
# (es. Eitch ha email-marketing attivo) — NON aggiungerli globalmente
}
# Prefissi skill di area (per identificare skill-fantasma)
AREA_SKILL_PREFIXES = ('mainstream-', 'pb-', 'maiker-', 'personale-', 'qh7-',
'eitch-', 'denny-', 'drone24-')
# Cartelle archivio dove skip fix (vestigia)
ARCHIVE_PATH_FRAGMENTS = ('/4_Archivio/', '/archive/', '/4-archives/')
def is_skipped(path: Path) -> bool:
return any(p in SKIP_DIRS for p in path.parts)
def resolve_link(source: Path, target: str) -> Path | None:
target = target.strip().split('#')[0].split('?')[0].replace('%20', ' ').strip()
if not target or target.startswith(('http://', 'https://', 'mailto:', '#', 'tel:')):
return None
try:
resolved = (source.parent / target).resolve()
resolved.relative_to(VAULT)
return resolved
except (ValueError, OSError):
return None
def build_file_index(all_md: list[Path]) -> dict:
"""basename → list of paths nel vault."""
idx = defaultdict(list)
for f in all_md:
idx[f.name].append(f)
return idx
def find_broken_links():
all_md = [f for f in VAULT.rglob('*.md') if not is_skipped(f.relative_to(VAULT))]
idx = build_file_index(all_md)
broken = [] # list of (source_file, link_text, target_str, line_num, line_content)
for f in all_md:
try:
lines = f.read_text(encoding='utf-8').split('\n')
except (UnicodeDecodeError, OSError):
continue
for lineno, line in enumerate(lines, 1):
for m in MD_LINK_LINE.finditer(line):
text = m.group(1)
target_str = m.group(2).strip()
resolved = resolve_link(f, target_str)
if resolved is None:
continue # external or anchor
if resolved.suffix != '.md':
continue
if not resolved.exists():
broken.append({
'source': f,
'line_num': lineno,
'line_content': line,
'link_text': text,
'target_str': target_str,
'target_resolved': resolved,
'target_name': Path(target_str.split('#')[0]).name,
})
return broken, idx, all_md
def try_safe_rename(broken: dict, idx: dict, source: Path) -> str | None:
"""Strategy A: se esiste esattamente 1 file con lo stesso basename, riscrivi."""
name = broken['target_name']
if not name.endswith('.md'):
return None
matches = idx.get(name, [])
if len(matches) != 1:
return None
target_file = matches[0]
# Calcola path relativo da source
try:
rel = Path('/') / target_file.relative_to(VAULT)
src_rel = Path('/') / source.relative_to(VAULT)
# Relative
import os
new_target = os.path.relpath(str(target_file), str(source.parent))
return new_target
except (ValueError, OSError):
return None
def try_readme_to_index(broken: dict) -> str | None:
"""Strategy B: README.md → INDEX.md se INDEX.md esiste nella stessa cartella puntata."""
target_str = broken['target_str'].split('#')[0].split('?')[0].strip().replace('%20', ' ')
if not target_str.endswith('README.md'):
return None
# Build path to same folder but INDEX.md
try:
resolved_dir = (broken['source'].parent / target_str).resolve().parent
index_candidate = resolved_dir / 'INDEX.md'
if index_candidate.exists():
new_target = target_str[:-len('README.md')] + 'INDEX.md'
return new_target
except (ValueError, OSError):
pass
return None
def try_decommissioned_service(broken: dict) -> bool:
"""Strategy C: link a servizio decommissionato + la linea è una bullet/row.
Returns True se la linea intera va rimossa."""
target_str = broken['target_str']
# Controlla se il path contiene uno dei servizi decommissionati come sotto-cartella
for svc in DECOMMISSIONED_SERVICES:
pattern = f'/{svc}/MEMORY.md'
if target_str.endswith(pattern) or target_str.endswith(f'/{svc}/'):
# È un link a servizio scheletro mai esistito
# Controlla che la linea sia bullet o row markdown (altrimenti skip)
line = broken['line_content'].strip()
if line.startswith(('-', '*', '|')) or line.startswith('- ['):
return True
return False
def is_phantom_skill(broken: dict) -> bool:
"""Strategy D: target è una skill di area con prefisso noto ma file non esiste."""
name = broken['target_name']
if not name.endswith('.md'):
return False
# Rimuovi estensione per check prefix
base = name[:-3]
return any(base.startswith(p) for p in AREA_SKILL_PREFIXES)
def is_in_archive(broken: dict) -> bool:
"""Strategy E: source file è in cartella archivio → skip fix."""
src_str = str(broken['source'])
return any(frag in src_str for frag in ARCHIVE_PATH_FRAGMENTS)
def apply_fixes(broken_links: list, idx: dict):
"""Apply fixes to files, grouped by source file."""
fixes_by_file = defaultdict(list) # file → list of (action, data)
for b in broken_links:
# Strategy E (early skip): file in archivio → no fix
if is_in_archive(b):
continue
# Strategy A
new_target = try_safe_rename(b, idx, b['source'])
if new_target:
fixes_by_file[b['source']].append({
'action': 'rename',
'line_num': b['line_num'],
'old_target': b['target_str'],
'new_target': new_target,
})
continue
# Strategy B
new_target = try_readme_to_index(b)
if new_target:
fixes_by_file[b['source']].append({
'action': 'rename',
'line_num': b['line_num'],
'old_target': b['target_str'],
'new_target': new_target,
})
continue
# Strategie C/D (invasive): saltate in --safe-only
if SAFE_ONLY:
continue
# Strategy C
if try_decommissioned_service(b):
fixes_by_file[b['source']].append({
'action': 'remove_line',
'line_num': b['line_num'],
'old_target': b['target_str'],
})
continue
# Strategy D: skill fantasma → rimuovi il [text](path) dal markdown
# lasciando solo il testo (backtick-wrapped se possibile)
if is_phantom_skill(b):
fixes_by_file[b['source']].append({
'action': 'unwrap_link',
'line_num': b['line_num'],
'old_target': b['target_str'],
'link_text': b['link_text'],
})
continue
# Apply to files
stats = {'renamed': 0, 'removed_lines': 0, 'unwrapped': 0,
'files_touched': 0, 'manual_review': 0, 'archive_skipped': 0,
'unsafe_skipped': 0}
for src, actions in fixes_by_file.items():
try:
lines = src.read_text(encoding='utf-8').split('\n')
except (UnicodeDecodeError, OSError):
continue
# Sort actions by line_num desc (for removing lines without shifting)
actions.sort(key=lambda x: -x['line_num'])
lines_to_remove = set()
for a in actions:
lineno = a['line_num']
if a['action'] == 'rename':
old = f'({a["old_target"]})'
new = f'({a["new_target"]})'
if lineno <= len(lines):
lines[lineno - 1] = lines[lineno - 1].replace(old, new)
stats['renamed'] += 1
elif a['action'] == 'remove_line':
lines_to_remove.add(lineno - 1)
stats['removed_lines'] += 1
elif a['action'] == 'unwrap_link':
# [text](path) → `text` (phantom skill placeholder)
text = a['link_text']
# Extract base name senza .md per cleaner output
base = text.strip('`')
if base.endswith('.md'):
base = base[:-3]
old = f'[{text}]({a["old_target"]})'
new = f'`{base}`'
if lineno <= len(lines):
lines[lineno - 1] = lines[lineno - 1].replace(old, new)
stats['unwrapped'] += 1
if lines_to_remove:
lines = [l for i, l in enumerate(lines) if i not in lines_to_remove]
if not DRY_RUN:
src.write_text('\n'.join(lines), encoding='utf-8')
stats['files_touched'] += 1
# Count archive skipped
stats['archive_skipped'] = sum(1 for b in broken_links if is_in_archive(b))
# Determine which links were handled
handled = set()
for b in broken_links:
key = (str(b['source']), b['line_num'], b['target_str'])
if is_in_archive(b):
continue # not handled, but skipped intentionally
if try_safe_rename(b, idx, b['source']):
handled.add(key)
elif try_readme_to_index(b):
handled.add(key)
elif SAFE_ONLY and (try_decommissioned_service(b) or is_phantom_skill(b)):
# in safe-mode i C/D non sono gestiti: restano in manual_review
stats['unsafe_skipped'] += 1
elif not SAFE_ONLY and try_decommissioned_service(b):
handled.add(key)
elif not SAFE_ONLY and is_phantom_skill(b):
handled.add(key)
manual = [b for b in broken_links
if not is_in_archive(b)
and (str(b['source']), b['line_num'], b['target_str']) not in handled]
stats['manual_review'] = len(manual)
return stats, manual
def main():
print(f"🔍 Scansione link rotti...")
broken, idx, all_md = find_broken_links()
print(f" Trovati {len(broken)} link rotti in {len(all_md)} file .md totali.")
print(f"\n🛠️ Applico fix...")
stats, manual = apply_fixes(broken, idx)
mode = 'DRY-RUN' if DRY_RUN else 'APPLY'
if SAFE_ONLY:
mode += ' (SAFE-ONLY: solo A+B)'
print(f"\n=== {mode} DONE ===")
print(f"Link rinominati (safe rename + README→INDEX): {stats['renamed']}")
print(f"Righe rimosse (servizi decommissionati): {stats['removed_lines']}")
print(f"Link unwrapped (skill-fantasma → placeholder): {stats['unwrapped']}")
print(f"Skippati unsafe (--safe-only, C/D rinviati): {stats['unsafe_skipped']}")
print(f"Skippati (in archivio, non si toccano): {stats['archive_skipped']}")
print(f"File modificati: {stats['files_touched']}")
print(f"Manual review necessari: {stats['manual_review']}")
if manual and len(manual) <= 30:
print(f"\n⚠️ LINK ROTTI DA FIXARE MANUALMENTE:")
print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
seen = set()
for b in manual:
key = (b['source'].name, b['target_str'])
if key in seen:
continue
seen.add(key)
auto_note = ''
if SAFE_ONLY and (try_decommissioned_service(b) or is_phantom_skill(b)):
auto_note = ' (auto-fixabile con run completo)'
try:
rel = b['source'].relative_to(VAULT)
print(f" {rel}:{b['line_num']}{auto_note}")
print(f" → {b['target_str']}")
except ValueError:
pass
if __name__ == '__main__':
main()
Lo script che proietta i numeri
Il lettore degli snapshot con il filtro che riconosce una tabella come valida solo se l’intestazione è quella giusta (così non conta due volte le tabelle di confronto), e la funzione di proiezione ancorata all’ultimo dato reale.
#!/usr/bin/env python3
"""build-kpi-series.py — Serie storiche KPI + forecast da snapshot-piattaforme.md.
Sorgente (read-only): 2_Aree/personal-brand/2_Aree/_strategia-generale/dati/
snapshot-piattaforme.md — append-only, blocchi datati (header con YYYY-MM-DD).
Output (cache RIGENERATA DA ZERO a ogni run — mai editare a mano):
_strategia-generale/dati/serie/substack.csv (date,followers,paid,arr_usd)
_strategia-generale/dati/serie/instagram.csv (date,followers)
_strategia-generale/dati/serie/linkedin.csv (date,followers)
_strategia-generale/dati/serie/tiktok.csv (date,followers)
_strategia-generale/dati/serie/trend.md (tabella trend + forecast)
Letto soltanto (append-only MANUALE, regola in skill aggiorna-lavoro step 2):
2_Aree/mainstream-agency/3_Risorse/knowledge/finance/serie/mrr.csv
Parsing: una pipe-table è considerata snapshot SOLO se l'header è
`| Piattaforma | Follower... |` (esclude per costruzione le tabelle Δ
multi-data dello storico). Bullet storico solo nella sezione "Storico mensile"
ancorati a inizio riga. Conflitto tabella vs bullet stessa data → vince la
tabella (warning). YouTube escluso dalle serie.
Forecast DETERMINISTICO: ancorato all'ultima data dato (mai a oggi) —
rate = Δ/giorni sugli ultimi ≤4 punti; stima milestone = data_ultimo +
(milestone − valore)/rate. Advisory, non promessa.
Trigger: i cicli lo lanciano quando snapshot-piattaforme.md è nel delta git.
Regole: conventions §"Serie KPI". Idempotente. Stdlib only. Flag: --dry-run.
"""
import csv
import io
import re
import sys
from datetime import date, timedelta
from pathlib import Path
VAULT = Path('<IL_TUO_VAULT>') # <-- cambia con il path della tua cartella di note
SOURCE = VAULT / '2_Aree/personal-brand/2_Aree/_strategia-generale/dati/snapshot-piattaforme.md'
SERIE_DIR = SOURCE.parent / 'serie'
MRR_CSV = VAULT / '2_Aree/mainstream-agency/3_Risorse/knowledge/finance/serie/mrr.csv'
DRY_RUN = '--dry-run' in sys.argv
CHANNELS = {'substack': 'substack', 'instagram': 'instagram', 'ig': 'instagram',
'linkedin': 'linkedin', 'tiktok': 'tiktok', 'youtube': 'youtube'}
MILESTONES = {'substack': [250, 500], 'substack_paid': [10, 20],
'instagram': [400, 500], 'linkedin': [200, 250]}
RE_DATE_HEADER = re.compile(r'^#{2,3} .*?(\d{4}-\d{2}-\d{2})')
RE_BULLET = re.compile(r'(\*\*)?(IG|Instagram|TikTok|LinkedIn|Substack|YouTube)(\*\*)?\s*:\s*([^·|]+)', re.IGNORECASE)
def extract_number(cell: str):
"""'**210 abb** (🎉 7 paid…)' → 210 · '~1.300 (1,3K)' → 1300 · '1,3K' → 1300."""
s = cell.replace('**', '').replace('~', '').strip()
m = re.search(r'(\d{1,3}(?:\.\d{3})+|\d+(?:,\d)?K|\d+)', s)
if not m:
return None
tok = m.group(1)
if tok.endswith('K'):
return int(float(tok[:-1].replace(',', '.')) * 1000)
return int(tok.replace('.', ''))
def parse_source():
"""→ {date_iso: {channel: {'followers': n, 'paid': n|None, 'arr': n|None, 'from': 'table'|'bullet'}}}"""
data = {}
cur_date = None
in_storico = False
lines = SOURCE.read_text(encoding='utf-8').split('\n')
i = 0
while i < len(lines):
line = lines[i]
m = RE_DATE_HEADER.match(line)
if m:
cur_date = m.group(1)
in_storico = bool(re.search(r'storico', line, re.IGNORECASE))
i += 1
continue
if line.startswith('## ') or line.startswith('### '):
in_storico = bool(re.search(r'storico', line, re.IGNORECASE)) or in_storico and not RE_DATE_HEADER.match(line)
# un header non datato dentro lo storico mantiene il flag; uno nuovo datato lo resetta sopra
# pipe-table snapshot (gate: header Piattaforma + Follower)
if cur_date and line.strip().startswith('|') and 'Piattaforma' in line and re.search(r'Follower', line, re.IGNORECASE):
i += 2 # salta header + separatore
while i < len(lines) and lines[i].strip().startswith('|'):
cells = [c.strip() for c in lines[i].strip().strip('|').split('|')]
if len(cells) >= 2:
ch = CHANNELS.get(cells[0].replace('**', '').strip().lower())
if ch and ch != 'youtube':
n = extract_number(cells[1])
if n is not None:
rec = data.setdefault(cur_date, {})
prev = rec.get(ch)
entry = {'followers': n, 'paid': None, 'arr': None, 'from': 'table'}
pm = re.search(r'(\d+)\s*paid', cells[1])
am = re.search(r'(\d+)\s*USD\s*ARR', cells[1])
if pm:
entry['paid'] = int(pm.group(1))
if am:
entry['arr'] = int(am.group(1))
if prev and prev['from'] == 'table' and prev['followers'] != n:
print(f"⚠️ conflitto tabella {cur_date}/{ch}: {prev['followers']} vs {n} (tengo il primo)")
elif prev and prev['from'] == 'bullet':
rec[ch] = entry # tabella vince
elif not prev:
rec[ch] = entry
i += 1
continue
# bullet storico (solo in sezione storico, ancorati a inizio riga '- ')
if cur_date and in_storico and line.lstrip().startswith('- '):
for bm in RE_BULLET.finditer(line):
ch = CHANNELS.get(bm.group(2).lower())
if not ch or ch == 'youtube':
continue
n = extract_number(bm.group(4))
if n is None:
continue
rec = data.setdefault(cur_date, {})
if ch in rec:
if rec[ch]['from'] == 'table' and rec[ch]['followers'] != n:
print(f"⚠️ conflitto {cur_date}/{ch}: tabella {rec[ch]['followers']} vs bullet {n} (vince tabella)")
continue
entry = {'followers': n, 'paid': None, 'arr': None, 'from': 'bullet'}
pm = re.search(r'(\d+)\s*paid', bm.group(4))
if pm:
entry['paid'] = int(pm.group(1))
rec[ch] = entry
i += 1
return data
def write_csv(path: Path, header: list, rows: list):
buf = io.StringIO()
w = csv.writer(buf, lineterminator='\n')
w.writerow(header)
w.writerows(rows)
content = buf.getvalue()
if DRY_RUN:
print(f'--- {path.name} ---')
print(content.strip())
else:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding='utf-8')
def load_series(path: Path):
"""CSV → [(date, value)] dalla colonna 2."""
if not path.exists():
return []
rows = []
for r in csv.reader(path.read_text(encoding='utf-8').splitlines()[1:]):
if len(r) >= 2 and r[0] and r[1]:
try:
rows.append((date.fromisoformat(r[0]), float(r[1])))
except ValueError:
continue
return sorted(rows)
def forecast(points, milestones):
"""→ (ultimo, ritmo_str, stime[(milestone, data_iso)]) — ancorato all'ultimo dato."""
if len(points) < 2:
return None
pts = points[-4:]
days = (pts[-1][0] - pts[0][0]).days or 1
rate = (pts[-1][1] - pts[0][1]) / days
ritmo = f'{rate:+.1f}/gg su {days}gg'
stime = []
if rate >= 0.1:
for m in milestones:
if m > pts[-1][1]:
eta = pts[-1][0] + timedelta(days=round((m - pts[-1][1]) / rate))
stime.append((m, eta))
return (pts[-1], ritmo, stime, rate)
def main():
if not SOURCE.exists():
print(f'🔴 sorgente mancante: {SOURCE}')
sys.exit(2)
data = parse_source()
if not data:
print('🔴 nessun dato parsato dalla sorgente')
sys.exit(2)
dates = sorted(data.keys())
last_date = dates[-1]
if not data[last_date]:
print(f'⚠️ ultimo snapshot {last_date} non ha prodotto righe: formato cambiato?')
# CSV per canale
for ch in ('substack', 'instagram', 'linkedin', 'tiktok'):
rows = []
for d in dates:
e = data[d].get(ch)
if not e:
continue
if ch == 'substack':
rows.append([d, e['followers'],
e['paid'] if e['paid'] is not None else '',
e['arr'] if e['arr'] is not None else ''])
else:
rows.append([d, e['followers']])
header = ['date', 'followers', 'paid', 'arr_usd'] if ch == 'substack' else ['date', 'followers']
write_csv(SERIE_DIR / f'{ch}.csv', header, rows)
print(f' {ch}.csv: {len(rows)} punti')
# trend.md
L = ['---', 'tipo: report', 'audience: human+llm', 'generato-da: build-kpi-series.py',
f'freshness: {last_date}',
'tldr: "Trend KPI da serie CSV (fonte: snapshot-piattaforme.md). Forecast run-rate ancorato all\'ultimo dato, deterministico. NON editare a mano."',
'---', '', '# 📈 Trend KPI', '',
f'**Dati fino a**: {last_date} · fonte: snapshot-piattaforme.md (+ mrr.csv manuale)', '',
'| Canale | Ultimo valore | Ritmo (ultimi ≤4 punti) | Prossima milestone | Stima |',
'|---|---:|---|---:|---|']
def add_row(label, points, milestones):
fc = forecast(points, milestones)
if not fc:
L.append(f'| {label} | (dati insufficienti) | — | — | — |')
return
(d, v), ritmo, stime, rate = fc
if stime:
ms, eta = stime[0]
L.append(f'| {label} | {v:.0f} ({d.day}/{d.month}) | {ritmo} | {ms} | ~{eta.day}/{eta.month} |')
else:
note = 'trend piatto/negativo' if rate < 0.1 else 'milestone raggiunte'
L.append(f'| {label} | {v:.0f} ({d.day}/{d.month}) | {ritmo} | — | {note} |')
sub = load_series(SERIE_DIR / 'substack.csv')
add_row('Substack', sub, MILESTONES['substack'])
# paid: colonna 3 del csv substack
paid_pts = []
p = SERIE_DIR / 'substack.csv'
if p.exists():
for r in csv.reader(p.read_text(encoding='utf-8').splitlines()[1:]):
if len(r) >= 3 and r[0] and r[2]:
paid_pts.append((date.fromisoformat(r[0]), float(r[2])))
add_row('Substack paid', sorted(paid_pts), MILESTONES['substack_paid'])
add_row('Instagram', load_series(SERIE_DIR / 'instagram.csv'), MILESTONES['instagram'])
add_row('LinkedIn', load_series(SERIE_DIR / 'linkedin.csv'), MILESTONES['linkedin'])
add_row('TikTok', load_series(SERIE_DIR / 'tiktok.csv'), [])
add_row('MRR Mainstream (€ nominale)', load_series(MRR_CSV), [])
L += ['', '> Metodo: rate = Δ/giorni sugli ultimi ≤4 punti; stima = data_ultimo_dato + (milestone − valore)/rate. **Advisory, non promessa.** mrr.csv è append-only manuale (regola in skill step 2).', '']
trend = '\n'.join(L)
if DRY_RUN:
print(trend)
else:
(SERIE_DIR / 'trend.md').write_text(trend, encoding='utf-8')
print(f'✅ scritto serie/ ({len(dates)} snapshot, ultimo {last_date}) + trend.md')
if __name__ == '__main__':
main()
Lo scheletro del registro delle scadenze
Il file dove il vault deposita le scadenze che tengo nel gestionale Agency OS via MCP, e da cui il radar le pesca. Una riga per evento, con la data in formato anno-mese-giorno e un trattino davanti quando è una stima. Se non usi un gestionale lo compili a mano: la struttura è identica.
| Data | Evento | Area/Cliente | Note |
|---|---|---|---|
| 2026-07-01 | Lancio milestone | canale | nota breve |
| ~2026-07-20 | Rinnovo contratto | cliente | stima, anticipare la conversazione |
I numeri grezzi dei primi giri
Radar: ventotto fonti reali al primo giro pulito, zero falsi positivi rimasti dopo aver tarato le liste nere. Riparatore: link rotti del vault scesi da centoventisei a novantaquattro nei primi interventi, nove sistemati nel primo giro in modalità sicura. Stato del vault: da un file fermo diciannove giorni a zero file fermi, una volta acceso il controllo sull’età.
Il punto che mi porto dietro da queste tre settimane è che non ho reso il mio second brain più intelligente. Gli ho tolto l’abitudine di aspettare. Lo stesso meccanismo che usavo per non far invecchiare le note, misurare l’età di una cosa e farla emergere quando supera una soglia, l’ho girato verso il futuro. Una scadenza, un link, un numero: ognuno ha la sua sentinella. E la sera, quando apro il vault, non parto più dalla paura di essermi perso qualcosa. Parto da una lista. Forse è merito di mia mamma che mi ha insegnato a non lasciare le cose a metà. Anche le note.
Se vuoi i prossimi script scaricabili e i sistemi completi, c’è il piano maikers paganti. Cinque euro al mese, niente attriti.
Tu come tieni vivo il tuo sistema di note: lo controlli a mano, hai qualche automazione, o ti fidi solo della memoria? Scrivimelo nei commenti, i casi migliori finiscono nel prossimo Second Brain.
Se ti è servito, condividilo con qualcuno che sta annegando nei suoi appunti. A volte il problema non è prendere più note, è farsele leggere al momento giusto.
Ivan Mosetti · Marketers di Giorno · Nerd AI di Notte · ivanmosetti.ai
PROMPT PER COSTRUIRE IL TUO RADAR A 30 GIORNI DOPO IL PAYWALL
Incolla questo a Claude (in Cowork o Claude Code, con accesso alla cartella del tuo vault) per farti costruire la prima versione del radar sul tuo sistema, adattata a come scrivi tu le date.
