Initial commit

This commit is contained in:
David Frassi
2026-06-17 01:26:44 +02:00
commit 51c56a6f30
65 changed files with 3731 additions and 0 deletions
+8
View File
@@ -0,0 +1,8 @@
GOOGLE_API_KEY=AIzaSyD6deyt0d3yp88sLbFaadJKxcOg-5cGy-A
BUFFER_ACCESS_TOKEN=EygpqeH802evGWSXqJ9EDSL_OpVBAFOBZqoVzw6OS9u
MAKE_WEBHOOK_URL=optional_make_webhook
TELEGRAM_BOT_TOKEN=your_bot_token
TELEGRAM_CHAT_ID=your_chat_id
# Higgsfield Credentials
HIGGSFIELD_API_KEY=792d35ce-5429-4056-a2b4-5b20bf74cf69
+5
View File
@@ -0,0 +1,5 @@
.env
.venv/
__pycache__/
*.pyc
.DS_Store
+25
View File
@@ -0,0 +1,25 @@
FROM python:3.11-slim
# Install system dependencies
RUN apt-get update && apt-get install -y \
ffmpeg \
libmagic1 \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Aggiungiamo src al PYTHONPATH così gli import funzionano da ovunque
ENV PYTHONPATH=/app/src
# Copy requirements and install
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy the rest of the application
COPY . .
# Expose port for Streamlit
EXPOSE 8501
# Avvio di Streamlit puntando alla nuova cartella src
CMD ["streamlit", "run", "src/app.py", "--server.port=8501", "--server.address=0.0.0.0"]
+43
View File
@@ -0,0 +1,43 @@
# 📊 Guida alla Sincronizzazione Real-Time dei Crediti Higgsfield AI
Questa guida spiega come abilitare il tracciamento automatico e in tempo reale del consumo crediti del tuo abbonamento Higgsfield AI (Pro Plan) all'interno della dashboard Streamlit.
Seguendo le tue indicazioni di sicurezza, **il sistema non effettua scansioni del tuo browser Chrome né accede al portachiavi di sistema (Chrome Safe Storage)**. La sincronizzazione si basa unicamente sul token di sessione configurato nel file `.env`.
---
## 🛠️ Cosa è stato fatto
1. **Automazione Completa**: Rimosso il form manuale per regolare il budget dei crediti dal cruscotto Streamlit. Ora i crediti si allineano da soli online.
2. **Aggiornamento REST Real-Time**: Implementato un client REST ufficiale (`HiggsClient`) in `src/app.py` che, ad ogni avvio o refresh del cruscotto, interroga in tempo reale i server di Higgsfield per mostrare la percentuale e i crediti rimanenti effettivi.
3. **Indicatori Grafici Premium**:
* **🟢 SINCRONIZZATO ONLINE**: Indica che il token è valido e i crediti mostrati provengono in diretta dal tuo account, con data e ora dell'ultimo allineamento.
* **🟡 CONTATORE LOCALE (CACHE)**: Compare se il token non è impostato nel file `.env`, indicandoti come procedere.
* **🔴 ERRORE DI CONNESSIONE**: Compare in caso di token scaduto o problemi temporanei di connessione, mantenendo visibile l'ultima cache per stabilità.
---
## 🔑 Come configurare la sincronizzazione automatica
Per sbloccare il monitoraggio automatico dei crediti reali, devi aggiungere il tuo token di sessione all'interno del file di configurazione ambientale. Segui questi passi semplici:
1. Apri il browser ed effettua l'accesso sul sito ufficiale **[higgsfield.ai](https://higgsfield.ai)**.
2. Premi `F12` (oppure clicca col tasto destro sulla pagina e seleziona **Ispeziona**) per aprire gli Strumenti per Sviluppatori del browser.
3. Spostati nella scheda **Application** (su Google Chrome) o **Storage** (su Firefox/Safari).
4. Nel menu laterale sinistro, espandi la voce **Cookies** e seleziona `https://higgsfield.ai`.
5. Cerca nell'elenco la riga con il nome **`__session`**.
6. Fai doppio clic sul valore del cookie per selezionarlo interamente e **copialo** (è una lunga stringa alfanumerica che inizia con `eyJ...`).
7. Apri il file [**.env**](file:///Users/davidfrassi/SRC/agenti/agenzia/.env) situato nella cartella principale del progetto.
8. Aggiungi la seguente variabile in fondo al file incollando il token copiato:
```env
HIGGSFIELD_TOKEN=eyJ...[incolla_qui_il_tuo_token]
```
9. Salva il file `.env`.
---
## 🚀 Risultato Atteso
Una volta salvata la variabile nel file `.env`, ricarica il cruscotto Streamlit. Il pannello in basso dedicato ai crediti:
* Rileverà automaticamente il token.
* Mostrerà il badge verde **🟢 SINCRONIZZATO ONLINE** con la data e l'ora dell'aggiornamento.
* Mostrerà i tuoi reali crediti rimanenti online e la percentuale esatta consumata, in perfetto allineamento con la dashboard ufficiale di Higgsfield!
+37
View File
@@ -0,0 +1,37 @@
# Roadmap Implementazione: Redazione Social Multi-Agente
Questo documento elenca i passi necessari per costruire il sistema di automazione social.
## Fase 1: Setup Ambiente e Dati (Dockerized)
- [ ] Inizializzare un progetto con **Docker** e **docker-compose**.
- [ ] Creare la struttura delle cartelle persistenti (volumi):
- `data/images/artista_n`: per le foto.
- `data/audio/artista_n`: per i brani .mp3 da analizzare.
- [ ] Creare un file `config.json` con i profili dei 4 cantanti e i link Spotify/Distrokid.
- [ ] Preparare il `Dockerfile` con le dipendenze (Python, FFmpeg per audio, ecc.).
## Fase 2: Sviluppo del Database delle Immagini
- [ ] Creare uno script che scansiona la cartella immagini.
- [ ] Implementare una logica di "tracking" per segnare le immagini già usate (per evitare ripetizioni).
- [ ] (Opzionale) Usare un modello Vision per pre-analizzare le immagini e salvarne una descrizione.
## Fase 3: Sviluppo degli Agenti
- [ ] **Agente Audio Analyst**: Utilizza modelli AI multimodali (es. Gemini) per ascoltare il brano .mp3 e dedurre genere, mood, bpm e temi principali.
- [ ] **Agente Asset Manager**: Seleziona l'immagine non usata dal folder specifico dell'artista.
- [ ] **Agente Copywriter**: Genera testi basandosi sull'output dell'Audio Analyst (non più su generi predefiniti).
- [ ] **Agente Visual-Harmonizer**: Incrocia l'analisi audio con l'estetica dell'immagine scelta.
- [ ] **Agente Hashtag**: Genera tag basati sul brano analizzato e sul visual.
- [ ] **Agente Supervisore**: Verifica finale della coerenza tra audio, immagine e link.
## Fase 4: Orchestrazione e Approvazione
- [ ] Configurare il grafo di **LangGraph** per collegare gli agenti.
- [ ] Creare una semplice interfaccia (es. **Streamlit**) per visualizzare la proposta del post e cliccare su "Approva" o "Rifiuta".
- [ ] Integrare un sistema di notifiche (es. Telegram Bot) per avvisarti quando un post è pronto.
## Fase 5: Distribuzione (Post reale)
- [ ] Integrare le API dei social media o un aggregatore (es. **Buffer API** o un webhook verso **Make.com**).
- [ ] Testare il flusso completo con un post di prova.
## Fase 6: Manutenzione
- [ ] Implementare un log dei post pubblicati.
- [ ] Sistema di alert se le immagini in una cartella stanno per finire.
+85
View File
@@ -0,0 +1,85 @@
{
"artists": [
{
"id": "Veronica Intorcia",
"name": "Veronica Intorcia",
"spotify_url": "https://open.spotify.com/intl-it/album/2LXeQfqTYA9SyYFz776pj5?si=k0caFWXaSHW63FLmrzifIg",
"distrokid_url": "https://distrokid.com/hyperfollow/veronika33/midnight-meridian-2",
"schedule_time": "09:00",
"drive_audio_url": "https://drive.google.com/file/d/1IfFSCPYuBBBCRaLerQYe30lrzcIjhqiI/view?usp=sharing",
"drive_images_url": "https://drive.google.com/drive/folders/1MlO5TkCGrnUNoNFzepM6NtnBnAjhyiHp?usp=drive_link",
"drive_videos_url": "https://drive.google.com/drive/folders/1oKRPFYZgusUtk0F9RWiC6a-Su8e4kZY4?usp=drive_link",
"buffer_token": "d0uh1tvuD4Y_VDJkVDlDDYZSGHOxVWDnJlLCvNoBiYO",
"narrative_style": "sile narrativo da cantnate pop figa stile dualipa",
"social_tag": "@veronika_officialpage",
"soul_id": "soul_veronica_intorcia",
"drive_starting_photos_url": "https://drive.google.com/drive/folders/1FAFyCaIInW5qGkv_pzviqLZKUbaSCIf-?usp=drive_link"
},
{
"id": "Cinzia Carreri",
"name": "Cinzia Carreri",
"spotify_url": "https://open.spotify.com/intl-it/track/0Z5xASrrUMx60kBXyqzTfe?si=f672e645783f4fd5",
"distrokid_url": "https://distrokid.com/hyperfollow/odette4/mirror-of-life-2",
"schedule_time": "10:30",
"drive_audio_url": "https://drive.google.com/file/d/1BVn0MlYzolR9HxwZGlsAC8YmIJGBbW8y/view?usp=drive_link",
"drive_images_url": "https://drive.google.com/drive/folders/1M7mD43vGKoBV_TsSjEDrN6GCsliQrvn1?usp=drive_link",
"drive_videos_url": "https://drive.google.com/drive/folders/1jB_BJH2QKh_XvrZz5qjTuDS65jQ_NJE6?usp=drive_link",
"buffer_token": "d0uh1tvuD4Y_VDJkVDlDDYZSGHOxVWDnJlLCvNoBiYO",
"narrative_style": "lo stile narrativo deve essere un po' sognatore da amanti del genere Enya e colonne sonore di film fantasy. Scrivi il post in inglese",
"social_tag": "@odette_officialpage",
"soul_id": "soul_cinzia_carreri"
},
{
"id": "Martina Zerjial",
"name": "Martina Zerjial",
"spotify_url": "https://open.spotify.com/intl-it/track/4jlDBu3SV24FUrH7YjTRzt?si=72142d6360e34370",
"distrokid_url": "https://distrokid.com/hyperfollow/elaya/saltwater-kisses",
"schedule_time": "14:00",
"drive_audio_url": "https://drive.google.com/file/d/13vZtqfVQ7G3Rjta1DvsYaAicEI1J__Ix/view?usp=drive_link",
"drive_images_url": "https://drive.google.com/drive/folders/1jkA5ApSNTJ68mEdHPmaVkBSa-CDJ0ihe?usp=drive_link",
"drive_videos_url": "https://drive.google.com/drive/folders/1Fk85iijYBAuznGMv_LZN018jvQOkH6M6?usp=drive_link",
"buffer_token": "d0uh1tvuD4Y_VDJkVDlDDYZSGHOxVWDnJlLCvNoBiYO",
"narrative_style": "stile narrativo da popstar che fa genere retrowave ma anche funky moderno, in stile the kolors",
"social_tag": "@elaya_officialpage",
"soul_id": "soul_martina_zerjial"
},
{
"id": "Ambra Manca",
"name": "Ambra Manca",
"spotify_url": "",
"distrokid_url": " ",
"schedule_time": "18:00",
"drive_audio_url": "https://drive.google.com/file/d/1ScgXJmnniTK8QeFZlB_qgCrywVaR5-_J/view?usp=drive_link",
"drive_images_url": "https://drive.google.com/drive/folders/1OXYnJNILkzVrlXNcMUbZBcKMrE9KT1ho?usp=drive_link",
"drive_videos_url": "",
"buffer_token": "d0uh1tvuD4Y_VDJkVDlDDYZSGHOxVWDnJlLCvNoBiYO",
"narrative_style": "stile narrativo figo da cantante funky americano, con contaminazione alla jamiroquai",
"social_tag": "@jaderaya_official",
"soul_id": "soul_ambra_manca"
},
{
"id": "StereoComics",
"name": "StereoComics",
"spotify_url": "",
"distrokid_url": "",
"schedule_time": "18:00",
"drive_audio_url": "https://drive.google.com/file/d/14RMoKsiupSQVj2MEzbXgNBzKZ3FPLSVQ/view?usp=sharing",
"drive_images_url": "https://drive.google.com/drive/folders/1aaReM9YqG8JhXMQKtZuFqWE9RbvwvsRp?usp=drive_link",
"drive_videos_url": "",
"buffer_token": "4OwhtCHLJ61Ld659qeQS1qd0MVFypTY3TtP1Wa9AKSA",
"narrative_style": "lo stile narrativo deve essere un po' canzonatorio, ironico, da nerd, che fa citazione degli aneddoti degli autori originari del brano, usa un linguaggio giovanile e accattivante",
"social_tag": "@spaziosigle",
"soul_id": "soul_stereocomics"
}
],
"settings": {
"daily_post_count": 1,
"platforms": [
"instagram",
"facebook",
"tiktok",
"x",
"youtube"
]
}
}
+51
View File
@@ -0,0 +1,51 @@
[2026-05-05 21:05:20] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:05:22] RISPOSTA BUFFER: {"post": {"id": "69fa5b92508055e61464dc15"}}
[2026-05-05 21:05:22] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:05:22] ERRORE HTTP 400: {"errors":[{"message":"Variable \"$input\" got invalid value { type: \"post\" } at \"input.metadata.instagram\"; Field \"shouldShareToFeed\" of required type \"Boolean!\" was not provided.","locations":[{"line":2,"column":23}],"extensions":{"code":"BAD_USER_INPUT"}}]}
[2026-05-05 21:07:05] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:07:06] RISPOSTA BUFFER: {"post": {"id": "69fa5bfaf2feba8c0c48e58f"}}
[2026-05-05 21:07:06] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:07:07] RISPOSTA BUFFER: {"post": {"id": "69fa5bfbf2feba8c0c48e5a9"}}
[2026-05-05 21:09:19] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:09:20] RISPOSTA BUFFER: {"post": {"id": "69fa5c80f2feba8c0c48e835"}}
[2026-05-05 21:09:20] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:09:21] RISPOSTA BUFFER: {"post": {"id": "69fa5c81508055e61464e273"}}
[2026-05-05 21:37:02] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:37:04] RISPOSTA BUFFER: {"post": {"id": "69fa6300f2feba8c0c492100"}}
[2026-05-05 21:37:04] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:37:05] RISPOSTA BUFFER: {"post": {"id": "69fa6301f2feba8c0c492127"}}
[2026-05-05 21:38:31] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:38:33] RISPOSTA BUFFER: {"post": {"id": "69fa6359508055e614652001"}}
[2026-05-05 21:38:33] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:38:35] RISPOSTA BUFFER: {"post": {"id": "69fa635bf2feba8c0c4922ab"}}
[2026-05-05 21:41:27] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:41:27] ERRORE HTTP 400: {"errors":[{"message":"Variable \"$input\" got invalid value { channelId: \"69f9e7e35c4c051afa116a9e\", text: \"*Un volo elettrico verso il nulla ✨*\\n\\nLuci notturne, riflessi dorati e il bisogno viscerale di lasciarsi tutto alle spalle. \\\"Midnight Meridian\\\" è la mia fuga urbana, una catarsi tra synth brillanti e ritmi che non lasciano scampo. Se senti la necessità di perderti nel movimento per ritrovare te stessa, questa è la tua nuova colonna sonora. Lascia che le ombre prendano vita e abbandonati a questa danza liberatoria. Ascolta ora: https://distrokid.com/hyperfollow/veronika33/midnight-meridian-2\\n\\n#MidnightMeridian #Veronika33 #Synthwave #ElectronicMusic #NewMusic #UrbanVibes #NightLife #Catharsis #Synthpop #DanceMusic #MusicRelease #MidnightVibes #IndependentArtist #ElectroPop #NightDrive\", mode: \"shareNow\", assets: { images: [Array] }, metadata: { tiktok: [Object] } }; Field \"schedulingType\" of required type \"SchedulingType!\" was not provided.","locations":[{"line":2,"column":23}],"extensions":{"code":"BAD_USER_INPUT"}}]}
[2026-05-05 21:41:27] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:41:27] ERRORE HTTP 400: {"errors":[{"message":"Variable \"$input\" got invalid value { channelId: \"69f9ea855c4c051afa117baa\", text: \"*Un volo elettrico verso il nulla ✨*\\n\\nLuci notturne, riflessi dorati e il bisogno viscerale di lasciarsi tutto alle spalle. \\\"Midnight Meridian\\\" è la mia fuga urbana, una catarsi tra synth brillanti e ritmi che non lasciano scampo. Se senti la necessità di perderti nel movimento per ritrovare te stessa, questa è la tua nuova colonna sonora. Lascia che le ombre prendano vita e abbandonati a questa danza liberatoria. Ascolta ora: https://distrokid.com/hyperfollow/veronika33/midnight-meridian-2\\n\\n#MidnightMeridian #Veronika33 #Synthwave #ElectronicMusic #NewMusic #UrbanVibes #NightLife #Catharsis #Synthpop #DanceMusic #MusicRelease #MidnightVibes #IndependentArtist #ElectroPop #NightDrive\", mode: \"shareNow\", assets: { images: [Array] }, metadata: { instagram: [Object] } }; Field \"schedulingType\" of required type \"SchedulingType!\" was not provided.","locations":[{"line":2,"column":23}],"extensions":{"code":"BAD_USER_INPUT"}}]}
[2026-05-05 21:42:30] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-05 21:42:41] RISPOSTA BUFFER: {"post": {"id": "69fa6447f2feba8c0c492ab3"}}
[2026-05-05 21:42:41] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-05 21:42:52] RISPOSTA BUFFER: {"post": {"id": "69fa6452f2feba8c0c492ae3"}}
[2026-05-06 00:27:00] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-06 00:27:01] ERRORE HTTP 400: {"errors":[{"message":"Variable \"$input\" got invalid value { video: { url: \"https://h.uguu.se/iBZbfOTe.mp4\" } } at \"input.assets\"; Field \"video\" is not defined by type \"AssetsInput\". [Suggestion hidden]?","locations":[{"line":2,"column":23}],"extensions":{"code":"BAD_USER_INPUT"}}]}
[2026-05-06 00:27:01] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-06 00:27:01] ERRORE HTTP 400: {"errors":[{"message":"Variable \"$input\" got invalid value { video: { url: \"https://h.uguu.se/iBZbfOTe.mp4\" } } at \"input.assets\"; Field \"video\" is not defined by type \"AssetsInput\". [Suggestion hidden]?","locations":[{"line":2,"column":23}],"extensions":{"code":"BAD_USER_INPUT"}}]}
[2026-05-06 00:28:05] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-06 00:28:19] RISPOSTA BUFFER: {"post": {"id": "69fa8b19e6fd3862f225c85c"}}
[2026-05-06 00:28:19] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-06 00:28:32] RISPOSTA BUFFER: {"post": {"id": "69fa8b25e6fd3862f225c8e1"}}
[2026-05-12 23:52:39] INVIO POST a Canale: 69f9e7e35c4c051afa116a9e
[2026-05-12 23:52:39] RISPOSTA BUFFER: {"message": "Channel not found"}
[2026-05-12 23:52:39] INVIO POST a Canale: 69f9ea855c4c051afa117baa
[2026-05-12 23:52:40] RISPOSTA BUFFER: {"message": "Channel not found"}
[2026-05-13 00:09:28] Nessun Profile ID fornito, avvio auto-discovery...
[2026-05-13 00:09:29] ERRORE recupero profili: {"errors":[{"message":"Cannot query field \"profiles\" on type \"Query\".","locations":[{"line":3,"column":11}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}
[2026-05-13 00:09:48] Nessun Profile ID fornito, avvio auto-discovery...
[2026-05-13 00:09:48] ERRORE recupero profili: {"errors":[{"message":"Cannot query field \"profiles\" on type \"Query\".","locations":[{"line":3,"column":11}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}]}
+5
View File
@@ -0,0 +1,5 @@
{
"total_monthly": 1000,
"remaining": 910,
"generated_this_month": 90
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 4.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.6 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.6 MiB

BIN
View File
Binary file not shown.
Binary file not shown.

After

Width:  |  Height:  |  Size: 1.3 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.4 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 478 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 312 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.3 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.8 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.7 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 6.0 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.2 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 578 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.0 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.3 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.2 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.9 MiB

+138
View File
@@ -0,0 +1,138 @@
--- Tue May 12 16:02:34 2026 ---
❌ Errore critico nel task generate (StereoComics): Failed to retrieve file url:
Cannot retrieve the public link of the file. You may need to change
the permission to 'Anyone with the link', or have had many accesses.
Check FAQ in https://github.com/wkentaro/gdown?tab=readme-ov-file#faq.
You may still be able to access the file from the browser:
https://drive.google.com/uc?id=14RMoKsiupSQVj2MEzbXgNBzKZ3FPLSVQ
but Gdown can't. Please check connections and permissions.
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/gdown/download.py", line 301, in download
url = get_url_from_gdrive_confirmation(res.text)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/gdown/download.py", line 65, in get_url_from_gdrive_confirmation
raise FileURLRetrievalError(
gdown.exceptions.FileURLRetrievalError: Cannot retrieve the public link of the file. You may need to change the permission to 'Anyone with the link', or have had many accesses. Check FAQ in https://github.com/wkentaro/gdown?tab=readme-ov-file#faq.
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/app/src/background_runner.py", line 71, in <module>
run_generate(args.artist_id, mode=args.mode)
File "/app/src/background_runner.py", line 43, in run_generate
results = agents.run_for_artist(artist, mode=mode)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/app/src/agents.py", line 285, in run_for_artist
audio_state = self.analyze_audio_node(initial_state)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/app/src/agents.py", line 49, in analyze_audio_node
gdown.download(url, output=path, quiet=True)
File "/usr/local/lib/python3.11/site-packages/gdown/download.py", line 312, in download
raise FileURLRetrievalError(message)
gdown.exceptions.FileURLRetrievalError: Failed to retrieve file url:
Cannot retrieve the public link of the file. You may need to change
the permission to 'Anyone with the link', or have had many accesses.
Check FAQ in https://github.com/wkentaro/gdown?tab=readme-ov-file#faq.
You may still be able to access the file from the browser:
https://drive.google.com/uc?id=14RMoKsiupSQVj2MEzbXgNBzKZ3FPLSVQ
but Gdown can't. Please check connections and permissions.
--- Tue May 12 20:58:46 2026 ---
❌ Errore critico nel task generate (Veronica Intorcia): (sqlite3.OperationalError) table drafts has no column named focus_points
[SQL: INSERT INTO drafts (artist_id, title, caption, hashtags, image_path, image_paths, image_url, video_url, video_path, audio_analysis, focus_points, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)]
[parameters: ('Veronica Intorcia', "Perdersi nell'Emozione", 'Lasciati trasportare dalle vibrazioni ipnotiche di "Lost In The Emotion". Una fusione magistrale tra Deep House e Melodic House, dove voci soul incon ... (151 characters truncated) ... ofisticata o vuole accendere l\'energia della notte. Ascolta ora il nuovo brano qui: https://distrokid.com/hyperfollow/veronika33/midnight-meridian-2', '#DeepHouse #MelodicHouse #ElectronicMusic #LostInTheEmotion #HouseMusic #SoulfulHouse #NewMusicAlert #HypnoticBeats #MusicJourney #NightVibes #MelodicTechno #IntrospectiveVibes #EuphoricMusic #DanceMusic #NewRelease', '', None, None, '', None, "Immergiti nelle profondità ipnotiche di 'Lost In The Emotion', una magistrale fusione di Deep House e Melodic House. Con le sue voci soul, il ritmo i ... (166 characters truncated) ... are un mood sofisticato o accendere l'energia notturna, è un viaggio sonoro accattivante, progettato per risuonare profondamente con gli ascoltatori.", None, 'pending', '2026-05-12 20:58:46.021420')]
(Background on this error at: https://sqlalche.me/e/20/e3q8)
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context
self.dialect.do_execute(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
sqlite3.OperationalError: table drafts has no column named focus_points
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/app/src/background_runner.py", line 72, in <module>
run_generate(args.artist_id, mode=args.mode)
File "/app/src/background_runner.py", line 46, in run_generate
db.save_draft(
File "/app/src/database.py", line 131, in save_draft
session.commit()
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 2030, in commit
trans.commit(_to_root=True)
File "<string>", line 2, in commit
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/state_changes.py", line 137, in _go
ret_value = fn(self, *arg, **kw)
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1311, in commit
self._prepare_impl()
File "<string>", line 2, in _prepare_impl
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/state_changes.py", line 137, in _go
ret_value = fn(self, *arg, **kw)
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1286, in _prepare_impl
self.session.flush()
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 4331, in flush
self._flush(objects)
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 4466, in _flush
with util.safe_reraise():
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/langhelpers.py", line 121, in __exit__
raise exc_value.with_traceback(exc_tb)
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 4427, in _flush
flush_context.execute()
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/unitofwork.py", line 466, in execute
rec.execute(self)
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/unitofwork.py", line 642, in execute
util.preloaded.orm_persistence.save_obj(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/persistence.py", line 93, in save_obj
_emit_insert_statements(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/persistence.py", line 1233, in _emit_insert_statements
result = connection.execute(
^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1419, in execute
return meth(
^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 527, in _execute_on_connection
return connection._execute_clauseelement(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1641, in _execute_clauseelement
ret = self._execute_context(
^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1846, in _execute_context
return self._exec_single_context(
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1986, in _exec_single_context
self._handle_dbapi_exception(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2363, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context
self.dialect.do_execute(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 952, in do_execute
cursor.execute(statement, parameters)
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) table drafts has no column named focus_points
[SQL: INSERT INTO drafts (artist_id, title, caption, hashtags, image_path, image_paths, image_url, video_url, video_path, audio_analysis, focus_points, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)]
[parameters: ('Veronica Intorcia', "Perdersi nell'Emozione", 'Lasciati trasportare dalle vibrazioni ipnotiche di "Lost In The Emotion". Una fusione magistrale tra Deep House e Melodic House, dove voci soul incon ... (151 characters truncated) ... ofisticata o vuole accendere l\'energia della notte. Ascolta ora il nuovo brano qui: https://distrokid.com/hyperfollow/veronika33/midnight-meridian-2', '#DeepHouse #MelodicHouse #ElectronicMusic #LostInTheEmotion #HouseMusic #SoulfulHouse #NewMusicAlert #HypnoticBeats #MusicJourney #NightVibes #MelodicTechno #IntrospectiveVibes #EuphoricMusic #DanceMusic #NewRelease', '', None, None, '', None, "Immergiti nelle profondità ipnotiche di 'Lost In The Emotion', una magistrale fusione di Deep House e Melodic House. Con le sue voci soul, il ritmo i ... (166 characters truncated) ... are un mood sofisticato o accendere l'energia notturna, è un viaggio sonoro accattivante, progettato per risuonare profondamente con gli ascoltatori.", None, 'pending', '2026-05-12 20:58:46.021420')]
(Background on this error at: https://sqlalche.me/e/20/e3q8)
--- Thu May 28 09:50:25 2026 ---
❌ Errore critico nel task higgsfield (Veronica Intorcia): name 'subprocess' is not defined
Traceback (most recent call last):
File "/Users/davidfrassi/SRC/agenti/agenzia/src/background_runner.py", line 136, in <module>
run_higgsfield(args.artist_id, args.prompt, dry_run=args.dry_run)
~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/davidfrassi/SRC/agenti/agenzia/src/background_runner.py", line 114, in run_higgsfield
res = subprocess.run(cmd, capture_output=True, text=True)
^^^^^^^^^^
NameError: name 'subprocess' is not defined. Did you forget to import 'subprocess'?
+21
View File
@@ -0,0 +1,21 @@
services:
web:
container_name: redazione_web
build: .
volumes:
- .:/app
ports:
- "8501:8501"
env_file:
- .env
command: streamlit run src/app.py --server.port=8501 --server.address=0.0.0.0
scheduler:
container_name: redazione_scheduler
build: .
volumes:
- .:/app
- ./videogenerati_test:/app/videogenerati
env_file:
- .env
command: python src/scheduler_service.py
+6
View File
@@ -0,0 +1,6 @@
docker stop $(docker ps -aq) 2>/dev/null; \
docker rm $(docker ps -aq) 2>/dev/null; \
docker rmi -f $(docker images -aq) 2>/dev/null; \
docker network prune -f; \
docker volume prune -f; \
docker system prune -a --volumes -f
+31
View File
@@ -0,0 +1,31 @@
import os
import requests
import json
from dotenv import load_dotenv
load_dotenv()
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
query = """
query {
__type(name: "PostInputMetaData") {
name
inputFields {
name
type {
name
kind
ofType {
name
kind
}
}
}
}
}
"""
resp = requests.post(url, headers=headers, json={'query': query})
print(json.dumps(resp.json(), indent=2))
+48
View File
@@ -0,0 +1,48 @@
#!/bin/bash
# ==============================================================================
# Script: list_soul_ids.sh
# Descrizione: Si collega a Higgsfield AI tramite il token predisposto in .env
# e scarica l'elenco dei Soul ID dei personaggi con il loro nome.
# ==============================================================================
# Cambia directory sul percorso del progetto
cd "$(dirname "$0")"
echo "===================================================================="
echo "⚡ Higgsfield Character Discovery Utility"
echo "===================================================================="
# 1. Carica le variabili dal file .env se presente
if [ -f .env ]; then
# Esporta le variabili escludendo i commenti
export $(grep -v '^#' .env | xargs)
echo "✅ File .env caricato con successo."
else
echo "⚠️ Attenzione: File .env non trovato. Verranno usate le variabili di sistema."
fi
# 2. Verifica la presenza di Python e dell'ambiente virtuale
if [ -f ".venv/bin/python3" ]; then
PYTHON_EXE=".venv/bin/python3"
echo "✅ Utilizzo dell'ambiente virtuale locale (.venv)."
else
PYTHON_EXE="python3"
echo "️ Ambiente virtuale locale non trovato. Utilizzo del python di sistema."
fi
# 3. Avvia lo script di recupero e formattazione
echo "📡 Connessione a Higgsfield AI in corso..."
echo ""
$PYTHON_EXE -m src.list_higgsfield_characters
EXIT_CODE=$?
echo ""
if [ $EXIT_CODE -eq 0 ]; then
echo "✅ Ricerca completata con successo."
else
echo "❌ Errore durante il recupero dei dati da Higgsfield."
fi
echo "===================================================================="
exit $EXIT_CODE
+14
View File
@@ -0,0 +1,14 @@
langchain
langgraph
langchain-google-genai
google-generativeai
streamlit
gdown
python-dotenv
Pillow
sqlalchemy
pydantic
ffmpeg-python
python-magic
python-telegram-bot
apscheduler
+343
View File
@@ -0,0 +1,343 @@
import os
import json
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from audio_analyzer import AudioAnalyzer
from database import Database
from dotenv import load_dotenv
import google.generativeai as genai
load_dotenv()
class AgentState(TypedDict):
artist_id: str
artist_name: str
spotify_url: str
distrokid_url: str
audio_analysis: str
image_path: str
image_url: str
video_url: str # NUOVO
image_description: str
climax_start_sec: int # NUOVO
bpm: int # NUOVO
title: str
caption: str
hashtags: str
review_status: str
narrative_style: str
social_tag: str # UNIFICATO
class SocialAgents:
def __init__(self):
self.db = Database()
self.analyzer = AudioAnalyzer()
genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
self.model = genai.GenerativeModel('gemini-3-flash-preview')
def analyze_audio_node(self, state: AgentState):
audio = self.db.get_audio_for_artist(state['artist_id'])
if audio:
path = audio.file_path
if not os.path.exists(path) and audio.source_url:
os.makedirs("data/cache", exist_ok=True)
import gdown
# Forza estensione mp3 se il path non ce l'ha
if not path.endswith(('.mp3', '.wav', '.ogg')):
path += ".mp3"
print(f"📥 Download audio da Drive: {audio.source_url}...")
url = f"https://drive.google.com/uc?id={audio.source_url}"
gdown.download(url, output=path, quiet=True)
else:
print(f"✅ Audio già in cache: {path}")
# Specifichiamo il mime_type se necessario per l'analyzer
raw_analysis = self.analyzer.analyze_audio(path)
# PULIZIA IMMEDIATA per risparmiare spazio
if os.path.exists(path): os.remove(path)
try:
# Usa regex per estrarre solo il blocco JSON
import re
match = re.search(r'\{.*\}', raw_analysis, re.DOTALL)
if match:
clean_json = match.group(0)
else:
clean_json = raw_analysis.replace("```json", "").replace("```", "").strip()
data = json.loads(clean_json)
return {
"audio_analysis": data.get("analysis", raw_analysis),
"climax_start_sec": data.get("climax_start_sec", 0),
"bpm": data.get("bpm", 120)
}
except:
return {"audio_analysis": raw_analysis, "climax_start_sec": 0, "bpm": 120}
return {"audio_analysis": "Nessun brano trovato.", "bpm": 120}
def select_image_node(self, state: AgentState):
image = self.db.get_unused_image(state['artist_id'])
if image:
path = image.file_path
# Se il file non esiste localmente ma abbiamo l'ID Drive, lo scarichiamo in cache
if not os.path.exists(path) and image.source_url:
os.makedirs("data/cache", exist_ok=True)
import gdown
print(f"📥 Scaricando in cache: {path}")
try:
url = f"https://drive.google.com/uc?id={image.source_url}"
gdown.download(url, output=path, quiet=True)
except Exception as e:
return {"caption": None, "error": f"Errore download immagine: {e}"}
else:
print(f"✅ Immagine già in cache: {path}")
if not os.path.exists(path):
return {"caption": None, "error": f"File non trovato in cache dopo il download: {path}"}
# Upload file to Gemini
img_data = genai.upload_file(path)
# Wait for processing
while img_data.state.name == "PROCESSING":
import time
time.sleep(2)
img_data = genai.get_file(img_data.name)
response = self.model.generate_content(["Descrivi questa immagine per un post social musicale.", img_data])
# Segna come usata per non ripeterla mai più
self.db.mark_as_used(image.id)
# PULIZIA IMMEDIATA per risparmiare spazio
if os.path.exists(path): os.remove(path)
return {"image_path": path, "image_description": response.text, "image_url": image.source_url}
return {"image_path": "", "image_description": "Nessuna immagine disponibile.", "image_url": None}
def write_caption_node(self, state: AgentState):
is_video = "video" if state.get("video_url") else "foto"
style_instructions = state.get("narrative_style", "scrivi un post accattivante")
prompt = f"""
Sei un Social Manager esperto per l'artista {state['artist_name']}.
Il tuo obiettivo è scrivere un post che spacca seguendo queste ISTRUZIONI DI STILE:
---
{style_instructions}
---
REQUISITI AGGIUNTIVI:
- Tag Social: Inserisci il SOCIAL TAG dell'artista all'interno del CORPO del post in modo naturale.
- Connessione: Collega ironicamente quello che si vede nella {is_video} con il testo del brano.
- Conoscenza Nerd: Se le istruzioni lo richiedono, inserisci aneddoti o curiosità plausibili sugli autori originali.
CONTESTO:
ARTISTA: {state['artist_name']}
SOCIAL TAG: {state.get('social_tag', '')}
ANALISI AUDIO (Brano): {state['audio_analysis']}
DESCRIZIONE VISIVA ({is_video}): {state['image_description']}
LINK DI DESTINAZIONE: {state['distrokid_url']}
FORMATO RISPOSTA (RESTITUISCI SOLO QUESTO):
TITOLO: [Titolo corto e accattivante]
CORPO: [Testo del post seguendo lo stile richiesto, termina con il link {state['distrokid_url']}]
"""
response = self.model.generate_content(prompt).text
# Parsing semplice
title = ""
body = ""
if "TITOLO:" in response and "CORPO:" in response:
parts = response.split("CORPO:")
title = parts[0].replace("TITOLO:", "").strip()
body = parts[1].strip()
else:
body = response # fallback
return {"title": title, "caption": body}
def generate_hashtags_node(self, state: AgentState):
prompt = f"Genera 15 hashtag per questo post: {state['caption']}. Rispondi SOLO con gli hashtag, senza testo aggiuntivo o consigli."
response = self.model.generate_content(prompt).text
# Rimuove eventuali testi residui fuori dagli hashtag
tags = [word for word in response.split() if word.startswith("#")]
return {"hashtags": " ".join(tags)}
def build_workflow(self):
workflow = StateGraph(AgentState)
workflow.add_node("analyze_audio", self.analyze_audio_node)
workflow.add_node("select_image", self.select_image_node)
workflow.add_node("write_caption", self.write_caption_node)
workflow.add_node("generate_hashtags", self.generate_hashtags_node)
workflow.add_node("select_mixed_assets", self.select_mixed_assets_node)
workflow.add_node("generate_video", self.generate_video_node)
workflow.set_entry_point("analyze_audio")
workflow.add_edge("analyze_audio", "select_mixed_assets")
workflow.add_edge("select_mixed_assets", "write_caption")
workflow.add_edge("write_caption", "generate_video")
workflow.add_edge("generate_video", "generate_hashtags")
workflow.add_edge("generate_hashtags", END)
return workflow.compile()
def select_mixed_assets_node(self, state: AgentState):
assets = self.db.get_mixed_assets(state['artist_id'], total=12)
if not assets:
return {"image_path": "", "image_description": "Nessun asset disponibile."}
# 1. Seleziona una copertina significativa (Preferibilmente immagine MAI USATA)
cover = next((a for a in assets if a.file_type == 'image' and not a.is_used), assets[0])
# Altre immagini
others = [a for a in assets if a.id != cover.id]
sorted_assets = [cover] + others
# Marcatura immediata della cover per non riusarla
self.db.mark_as_used(cover.id)
asset_paths = []
os.makedirs("data/cache", exist_ok=True)
import gdown
for asset in sorted_assets:
path = asset.file_path
if not os.path.exists(path) and asset.source_url:
try:
url = f"https://drive.google.com/uc?id={asset.source_url}"
gdown.download(url, output=path, quiet=True)
asset_paths.append(path)
except: continue
elif os.path.exists(path):
asset_paths.append(path)
# 3. AI DESCRIZIONE E FACE DETECTION per Pan-Zoom intelligente
focus_points = []
desc = "Un montaggio video dinamico."
try:
print(f"👁️ AI: Ricerca volti (Pinpoint) e descrizione per {len(asset_paths)} asset...")
image_files = []
for p in asset_paths:
if p.lower().endswith(('.jpg', '.jpeg', '.png')):
image_files.append(genai.upload_file(p))
if image_files:
# Chiediamo sia la descrizione che le coordinate in un colpo solo
prompt = "Analyze these images for a high-end music video. " \
"1. For the first image (the cover), give a short poetic description. " \
"2. For ALL images, identify the EXACT center coordinates (x, y as percentage 0-100) of the main subject's face. " \
"If multiple people are present, pick the most attractive girl. " \
"Return ONLY JSON: {'description': '...', 'focus': [{'x': val, 'y': val}, ...]}"
response = self.model.generate_content(image_files + [prompt])
clean_json = response.text.replace('```json', '').replace('```', '').strip()
data = json.loads(clean_json)
desc = data.get('description', desc)
focus_points = data.get('focus', [])
print(f"✅ AI: Analisi completata. Focus points trovati: {len(focus_points)}")
except Exception as e:
print(f"⚠️ AI Regia fallita (uso default): {e}")
focus_points = [None] * len(asset_paths)
return {
"image_path": json.dumps(asset_paths),
"focus_points": json.dumps(focus_points),
"image_description": desc,
"image_url": cover.source_url
}
def generate_video_node(self, state: AgentState):
from video_generator import VideoGenerator
from gdown import download
audio = self.db.get_audio_for_artist(state['artist_id'])
if audio and state['image_path']:
vg = VideoGenerator()
audio_path = audio.file_path
if not os.path.exists(audio_path):
download(id=audio.source_url, output=audio_path, quiet=True)
try:
# Se image_path è una lista JSON (caso video)
try:
image_paths = json.loads(state['image_path'])
except:
image_paths = [state['image_path']]
print(f"🎬 AVVIO MONTAGGIO VIDEO: {len(image_paths)} asset + {audio_path} (Start: {state['climax_start_sec']}s, BPM: {state.get('bpm', 120)})")
focus_points = json.loads(state.get('focus_points', '[]'))
local_video = vg.generate_video(image_paths, audio_path, state['title'], start_time=state['climax_start_sec'], bpm=state.get('bpm', 120), focus_points=focus_points)
print(f"🚀 Caricamento video su host temporaneo...")
video_url = vg.upload_to_ephemeral(local_video)
print(f"✅ Video caricato: {video_url}")
# PULIZIA IMMEDIATA per risparmiare spazio
if os.path.exists(local_video): os.remove(local_video)
# La miniatura locale la teniamo solo se serve a Streamlit, ma qui la cancelliamo
# perché abbiamo detto No-Space e useremo l'URL cloud se possibile.
# In realtà vg.generate_thumbnail crea un .jpg che potremmo voler tenere,
# ma se l'utente vuole zero spazio, cancelliamo tutto.
return {"video_url": video_url, "video_path": local_video}
except Exception as e:
print(f"❌ ERRORE CRITICO GENERAZIONE VIDEO: {e}")
finally:
# PULIZIA ASSET ORIGINALI (Sempre, anche se fallisce)
try:
image_paths = json.loads(state['image_path'])
for p in image_paths:
if os.path.exists(p): os.remove(p)
except: pass
return {"video_url": ""}
def run_for_artist(self, artist_data, mode="both"):
"""Esegue i workflow. Mode: 'photo', 'video', 'both'"""
results = []
# 1. Analisi Audio (comune)
initial_state = {
"artist_id": artist_data['id'],
"artist_name": artist_data['name'],
"spotify_url": artist_data['spotify_url'],
"distrokid_url": artist_data['distrokid_url'],
"audio_analysis": "",
"image_path": "",
"image_url": "",
"video_url": "",
"image_description": "",
"climax_start_sec": 0,
"bpm": 120,
"title": "",
"caption": "",
"hashtags": "",
"review_status": "pending",
"narrative_style": artist_data.get('narrative_style', ''),
"social_tag": artist_data.get('social_tag', '')
}
audio_state = self.analyze_audio_node(initial_state)
initial_state.update(audio_state)
# 2. Generazione BOZZA FOTO
if mode in ["photo", "both"]:
print(f"📸 Avvio Workflow FOTO per {artist_data['name']}...")
photo_state = initial_state.copy()
photo_state.update(self.select_image_node(photo_state))
photo_state.update(self.write_caption_node(photo_state))
photo_state.update(self.generate_hashtags_node(photo_state))
results.append(photo_state)
# 3. Generazione BOZZA VIDEO
if mode in ["video", "both"]:
print(f"🎬 Avvio Workflow VIDEO per {artist_data['name']}...")
video_state = initial_state.copy()
video_state.update(self.select_mixed_assets_node(video_state))
video_state.update(self.write_caption_node(video_state))
video_state.update(self.generate_video_node(video_state))
video_state.update(self.generate_hashtags_node(video_state))
results.append(video_state)
return results
+643
View File
@@ -0,0 +1,643 @@
import streamlit as st
import json
import os
import sys
import subprocess
import glob
import signal
from database import Database
from buffer_publisher import BufferPublisher
from PIL import Image
from dotenv import load_dotenv
# Carichiamo le variabili d'ambiente (.env) prima di tutto
load_dotenv()
st.set_page_config(page_title="Redazione Social Multi-Agente", layout="wide")
# Inizializziamo il database e il publisher
db = Database()
publisher = BufferPublisher()
# Carichiamo la configurazione
with open('config.json', 'r') as f:
config = json.load(f)
artist_names = {a['id']: a['name'] for a in config['artists']}
artist_ids = list(artist_names.keys())
def load_credits_cache():
# Valori di default
cache_data = {"total_monthly": 1000, "remaining": 910, "generated_this_month": 90, "last_sync_status": "no_token"}
credits_file = "data/cache/credits.json"
try:
os.makedirs("data/cache", exist_ok=True)
if os.path.exists(credits_file):
with open(credits_file, 'r') as f:
loaded = json.load(f)
if isinstance(loaded, dict):
cache_data.update(loaded)
except:
pass
# Se c'è HIGGSFIELD_TOKEN nel file .env, facciamo la sincronizzazione in tempo reale online
token = os.getenv("HIGGSFIELD_TOKEN")
if token and token.strip() and token.startswith("eyJ"):
import datetime
try:
from higgsfield_cli.client import HiggsClient
client = HiggsClient(token=token)
wallet = client.get_wallet()
total = wallet.total_credits
remaining = int(wallet.credits_display)
generated = max(0, total - remaining)
now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cache_data.update({
"total_monthly": total,
"remaining": remaining,
"generated_this_month": generated,
"last_sync_status": "success",
"last_sync_time": now_str,
"sync_error": None
})
save_credits_cache(cache_data)
except Exception as e:
now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cache_data.update({
"last_sync_status": "failed",
"last_sync_time": now_str,
"sync_error": str(e)
})
# Non sovrascriviamo i valori reali passati in caso di errore di rete/timeout temporaneo, aggiorniamo solo lo stato del sync
save_credits_cache(cache_data)
else:
cache_data["last_sync_status"] = "no_token"
return cache_data
def save_credits_cache(data):
try:
os.makedirs("data/cache", exist_ok=True)
credits_file = "data/cache/credits.json"
with open(credits_file, 'w') as f:
json.dump(data, f, indent=2)
except:
pass
def get_running_tasks():
"""Ritorna la lista dei task attualmente in corso (basandosi sui file .lock)"""
locks = glob.glob("data/tasks/*.lock")
tasks = []
for lock in locks:
name = os.path.basename(lock).replace(".lock", "")
tasks.append(name)
return tasks
def create_manual_lock(task_name, artist_id="all"):
"""Crea un file di lock manualmente per feedback immediato nella UI"""
os.makedirs("data/tasks", exist_ok=True)
lock_file = f"data/tasks/{task_name}_{artist_id.replace(' ', '_')}.lock"
with open(lock_file, "w") as f:
f.write("starting")
return lock_file
def manage_task(task_id, action):
"""Gestisce un task in background: stop, restart o delete lock"""
lock_path = f"data/tasks/{task_id}.lock"
if not os.path.exists(lock_path):
return
pid_str = open(lock_path).read().strip()
# 1. Stop Process if exists
if pid_str.isdigit():
pid = int(pid_str)
try:
os.kill(pid, signal.SIGTERM)
print(f"Inviato SIGTERM a PID {pid}")
except ProcessLookupError:
pass
except Exception as e:
st.error(f"Errore chiusura processo: {e}")
# 2. Perform Action
if action == "stop" or action == "remove":
if os.path.exists(lock_path):
os.remove(lock_path)
if action == "restart":
# Determina i parametri dal task_id (es: generate_Ambra_Manca)
parts = task_id.split("_")
task_type = parts[0] # sync o generate
artist_id = "_".join(parts[1:]) if len(parts) > 1 else "all"
artist_id = artist_id.replace("_", " ") # Ripristina spazi
if os.path.exists(lock_path):
os.remove(lock_path)
create_manual_lock(task_type, artist_id)
cmd = [sys.executable, "src/background_runner.py", "--task", task_type]
if artist_id != "all":
cmd += ["--artist_id", artist_id]
subprocess.Popen(cmd)
# ============================================================
# SIDEBAR: Selezione Artista + Configurazione
# ============================================================
st.sidebar.header("🎤 Seleziona Artista")
selected_artist_idx = st.sidebar.selectbox(
"Artista",
options=range(len(config['artists'])),
format_func=lambda i: config['artists'][i]['name'],
label_visibility="collapsed"
)
# Rilevamento cambio artista per pulizia stato (Previene ghosting)
if "last_artist_idx" not in st.session_state:
st.session_state.last_artist_idx = selected_artist_idx
if st.session_state.last_artist_idx != selected_artist_idx:
st.session_state.last_artist_idx = selected_artist_idx
# Pulizia totale per evitare che i checkbox o le foto rimangano in trasparenza
st.session_state.clear()
st.rerun()
selected_artist = config['artists'][selected_artist_idx]
selected_artist_id = selected_artist['id']
selected_artist_name = selected_artist['name']
st.sidebar.markdown("---")
# Pannello di Configurazione
with st.sidebar.expander(f"⚙️ Configurazione {selected_artist_name}", expanded=False):
with st.form(key=f"form_edit_{selected_artist_idx}"):
new_name = st.text_input("Nome Artista", selected_artist.get('name', ''))
new_spotify = st.text_input("Spotify URL", selected_artist.get('spotify_url', ''))
new_distrokid = st.text_input("DistroKid URL", selected_artist.get('distrokid_url', ''))
new_time = st.text_input("Orario Pubblicazione (es. 09:00)", selected_artist.get('schedule_time', ''))
new_audio = st.text_input("Cartella Drive Audio (URL)", selected_artist.get('drive_audio_url', ''))
new_img = st.text_input("Cartella Drive Immagini (URL)", selected_artist.get('drive_images_url', ''))
new_vid = st.text_input("Cartella Drive Video (URL)", selected_artist.get('drive_videos_url', ''))
new_buffer_token = st.text_input("Buffer Access Token (Specifico)", selected_artist.get('buffer_token', ''), type="password")
new_style = st.text_area("Stile Narrativo (Istruzioni AI)", selected_artist.get('narrative_style', ""), height=100)
new_social_tag = st.text_input("Social Tag (es. @artista)", selected_artist.get('social_tag', ''))
new_soul_id = st.text_input("Higgsfield Soul ID 2.0", selected_artist.get('soul_id', ''))
new_starting_photos = st.text_input("Cartella Drive Foto Partenza / Modelli (URL)", selected_artist.get('drive_starting_photos_url', ''))
if st.form_submit_button("💾 Salva Modifiche"):
config['artists'][selected_artist_idx]['name'] = new_name
config['artists'][selected_artist_idx]['spotify_url'] = new_spotify
config['artists'][selected_artist_idx]['distrokid_url'] = new_distrokid
config['artists'][selected_artist_idx]['schedule_time'] = new_time
config['artists'][selected_artist_idx]['drive_audio_url'] = new_audio
config['artists'][selected_artist_idx]['drive_images_url'] = new_img
config['artists'][selected_artist_idx]['drive_videos_url'] = new_vid
config['artists'][selected_artist_idx]['buffer_token'] = new_buffer_token
config['artists'][selected_artist_idx]['narrative_style'] = new_style
config['artists'][selected_artist_idx]['social_tag'] = new_social_tag
config['artists'][selected_artist_idx]['soul_id'] = new_soul_id
config['artists'][selected_artist_idx]['drive_starting_photos_url'] = new_starting_photos
# Rimuovo i vecchi campi se esistenti per pulizia
for old_key in ['instagram_tag', 'tiktok_tag', 'instagram_profile_id', 'tiktok_profile_id']:
config['artists'][selected_artist_idx].pop(old_key, None)
with open('config.json', 'w') as f:
json.dump(config, f, indent=2)
st.success(f"Configurazione di {new_name} salvata!")
st.rerun()
st.sidebar.markdown("---")
running_tasks = get_running_tasks()
if running_tasks:
st.sidebar.warning("⏳ Task in background:")
for task_id in running_tasks:
col_t, col_stop, col_rel, col_del = st.sidebar.columns([4, 1, 1, 1])
col_t.caption(f"{task_id.replace('_', ' ')}")
if col_stop.button("🛑", key=f"stop_{task_id}", help="Ferma"):
manage_task(task_id, "stop")
st.rerun()
if col_rel.button("🔄", key=f"rel_{task_id}", help="Riavvia"):
manage_task(task_id, "restart")
st.rerun()
if col_del.button("🗑️", key=f"del_{task_id}", help="Rimuovi Lock"):
manage_task(task_id, "remove")
st.rerun()
if st.sidebar.button("🔄 Sincronizza Asset da Drive"):
if any(t.startswith("sync") for t in running_tasks):
st.sidebar.error("Sincronizzazione già in corso!")
else:
create_manual_lock("sync")
subprocess.Popen([sys.executable, "src/background_runner.py", "--task", "sync"])
st.sidebar.success("✅ Sincronizzazione avviata in background!")
st.rerun()
if st.sidebar.button("🗑️ Svuota Cache Locale"):
import shutil
try:
shutil.rmtree("data/cache")
os.makedirs("data/cache", exist_ok=True)
st.sidebar.success("✅ Cache svuotata con successo!")
st.session_state.clear()
st.rerun()
except Exception as e:
st.sidebar.error(f"Errore durante la pulizia: {e}")
# ============================================================
# AREA PRINCIPALE: Generazione e Gestione Bozze
# ============================================================
with st.container(key=f"main_content_{selected_artist_id}"):
st.title(f"🚀 Redazione: {selected_artist_name}")
# Sezione Generazione Rapida
st.subheader("✨ Nuova Generazione")
col_gen1, col_gen2 = st.columns(2)
gen_task_id_photo = f"generate_{selected_artist_id.replace(' ', '_')}_photo"
gen_task_id_video = f"generate_{selected_artist_id.replace(' ', '_')}_video"
# 1. Post Immagine (Leggero)
if gen_task_id_photo in running_tasks:
col_gen1.info("📸 Post Foto in corso...")
elif col_gen1.button("📸 Post Immagine (Solo 1 Foto)", use_container_width=True, key=f"btn_p_{selected_artist_id}"):
create_manual_lock("generate", f"{selected_artist_id}_photo")
subprocess.Popen([sys.executable, "src/background_runner.py", "--task", "generate", "--artist_id", selected_artist_id, "--mode", "photo"])
st.rerun()
# 2. Video Mix (Più pesante)
if gen_task_id_video in running_tasks:
col_gen2.info("🎬 Video Mix in corso...")
elif col_gen2.button("🎬 Video Mix (12 Asset)", use_container_width=True, key=f"btn_v_{selected_artist_id}"):
create_manual_lock("generate", f"{selected_artist_id}_video")
subprocess.Popen([sys.executable, "src/background_runner.py", "--task", "generate", "--artist_id", selected_artist_id, "--mode", "video"])
st.rerun()
st.markdown("---")
st.subheader("🎨 Generazione Foto Higgsfield AI (Soul ID)")
gen_task_id_higgsfield = f"higgsfield_{selected_artist_id.replace(' ', '_')}_both"
col_h1, col_h2 = st.columns([3, 1])
hf_prompt = col_h1.text_input("Prompt Stile Estetico", "A beautiful editorial studio portrait, highly detailed, cinematic studio lighting, professional photography, 8k resolution, crisp details", key=f"hf_p_{selected_artist_id}")
hf_dry_run = col_h2.checkbox("Simulazione (Dry-run)", value=False, key=f"hf_dr_{selected_artist_id}")
if gen_task_id_higgsfield in running_tasks:
st.info("🎨 Generazione Higgsfield AI in corso in background...")
else:
if st.button("🚀 Avvia Generazione Higgsfield AI per tutti i modelli", use_container_width=True, type="primary", key=f"btn_hf_{selected_artist_id}"):
# Check parameters
m_url = selected_artist.get("drive_starting_photos_url")
if not m_url:
st.error("Errore: Assicurati di aver configurato la 'Cartella Foto Partenza' nella barra laterale!")
else:
create_manual_lock("higgsfield", f"{selected_artist_id}_both")
cmd = [
sys.executable, "src/background_runner.py",
"--task", "higgsfield",
"--artist_id", selected_artist_id,
"--prompt", hf_prompt
]
if hf_dry_run:
cmd.append("--dry_run")
subprocess.Popen(cmd)
st.success("✅ Generazione Higgsfield avviata in background!")
st.rerun()
st.markdown("---")
# ============================================================
# BOZZE PENDENTI (filtrate per artista selezionato)
# ============================================================
st.subheader("📝 Bozze Pendenti")
drafts = db.get_pending_drafts(selected_artist_id)
if not drafts:
st.info(f"Nessuna bozza per {selected_artist_name}. Clicca su 'Genera Nuovo Post' per iniziare.")
else:
# Callback per seleziona/deseleziona tutto
if "current_draft_ids" not in st.session_state:
st.session_state.current_draft_ids = []
st.session_state.current_draft_ids = [d.id for d in drafts]
def toggle_all_drafts():
val = st.session_state.select_all_drafts_chk
for d_id in st.session_state.current_draft_ids:
st.session_state[f"sel_{d_id}"] = val
# Pre-calcolo delle bozze selezionate per evitare NameError nelle azioni bulk
selected_drafts = [d for d in drafts if st.session_state.get(f"sel_{d.id}", False)]
# Azioni di Gruppo
with st.expander("🛠️ Azioni di Gruppo", expanded=False):
col_btn1, col_btn2 = st.columns(2)
if col_btn1.button("🚀 PUBBLICA TUTTI I SELEZIONATI"):
if selected_drafts:
with st.spinner(f"Pubblicazione di {len(selected_drafts)} post..."):
for d in selected_drafts:
t_val = st.session_state.get(f"t_{d.id}", d.title)
c_val = st.session_state.get(f"c_{d.id}", d.caption)
h_val = st.session_state.get(f"h_{d.id}", d.hashtags)
db.update_draft(d.id, title=t_val, caption=c_val, hashtags=h_val)
full_text = f"*{t_val}*\n\n{c_val}\n\n{h_val}"
# Recupera dati specifici artista (Token e Profile IDs)
artist_data = next((a for a in config['artists'] if a['id'] == d.artist_id), None)
artist_token = artist_data.get("buffer_token") if artist_data else None
# Passiamo profile_ids vuoti per attivare l'auto-discovery nel publisher
profile_ids = []
publisher.publish(profile_ids, full_text, d.image_path, d.video_url, d.image_paths, artist_token=artist_token)
db.mark_as_published(d.id)
st.success(f"Pubblicati {len(selected_drafts)} post!")
st.rerun()
if col_btn2.button("🗑️ ELIMINA TUTTI I SELEZIONATI"):
if selected_drafts:
for d in selected_drafts:
db.delete_draft(d.id)
st.warning(f"Eliminati {len(selected_drafts)} post.")
st.rerun()
st.checkbox("Seleziona Tutte le Bozze", key="select_all_drafts_chk", on_change=toggle_all_drafts)
for draft in drafts:
with st.expander(f"📌 {draft.title or 'Bozza senza titolo'}", expanded=len(drafts) == 1):
c1, c2 = st.columns([1, 2])
with c1:
st.checkbox("Seleziona per Azione di Gruppo", key=f"sel_{draft.id}")
# ANTEPRIMA VIDEO (Priorità assoluta se esiste)
if draft.video_url:
st.video(draft.video_url)
st.success("🎬 Video 'CapCut-style' generato!")
# Anteprima Immagine (Metodo Base64 Infallibile)
display_url = draft.image_url
if not display_url and draft.image_path:
m_rec = db.get_media_by_path(draft.image_path)
if m_rec: display_url = m_rec.source_url
if display_url:
try:
import requests, base64
# Usiamo un URL di Drive che restituisce direttamente l'immagine
thumb_url = f"https://drive.google.com/thumbnail?id={display_url}&sz=w600"
response = requests.get(thumb_url, timeout=5)
if response.status_code == 200:
b64_img = base64.b64encode(response.content).decode()
st.markdown(f'<img src="data:image/jpeg;base64,{b64_img}" style="width:100%; border-radius: 10px; margin-bottom: 10px;">', unsafe_allow_html=True)
else:
# Fallback se il server non raggiunge Google
st.markdown(f'<img src="https://lh3.googleusercontent.com/u/0/d/{display_url}=w600" style="width:100%; border-radius: 10px;">', unsafe_allow_html=True)
except:
st.warning("🖼️ Errore caricamento (Cloud)")
elif draft.image_path and os.path.exists(draft.image_path):
try:
import base64
with open(draft.image_path, "rb") as img_file:
b64_img = base64.b64encode(img_file.read()).decode()
st.markdown(f'<img src="data:image/jpeg;base64,{b64_img}" style="width:100%; border-radius: 10px; margin-bottom: 10px;">', unsafe_allow_html=True)
except Exception as e:
st.image(draft.image_path)
elif draft.image_paths:
try:
paths = json.loads(draft.image_paths)
m_rec = db.get_media_by_path(paths[0])
if m_rec and m_rec.source_url:
img_url = f"https://lh3.googleusercontent.com/u/0/d/{m_rec.source_url}=w600"
st.markdown(f'<img src="{img_url}" style="width:100%; border-radius: 10px;">', unsafe_allow_html=True)
except: pass
elif not draft.video_url:
st.warning("⚠️ Anteprima non disponibile")
with c2:
edited_title = st.text_input("Titolo", draft.title, key=f"t_{draft.id}")
edited_caption = st.text_area("Testo", draft.caption, height=150, key=f"c_{draft.id}")
edited_tags = st.text_input("Hashtag", draft.hashtags, key=f"h_{draft.id}")
if draft.audio_analysis:
with st.expander("🎵 Pitch per Playlist Spotify (Generato dall'AI)", expanded=True):
st.info(draft.audio_analysis)
if st.button("🤖 Autocompleta Testi e Pitch con AI (Ascolta Audio)", key=f"ai_{draft.id}", use_container_width=True):
with st.spinner("Ascolto audio e generazione testi in corso..."):
from agents import SocialAgents
agents = SocialAgents()
audio = db.get_audio_for_artist(draft.artist_id)
if not audio:
st.error("Nessun audio trovato per questo artista.")
else:
state = {"artist_id": draft.artist_id, "artist_name": artist_names.get(draft.artist_id), "distrokid_url": ""}
if artist_data:
state["distrokid_url"] = artist_data.get("distrokid_url", "")
state["narrative_style"] = artist_data.get("narrative_style", "")
state["social_tag"] = artist_data.get("social_tag", "")
analysis_res = agents.analyze_audio_node(state)
state.update(analysis_res)
state["image_description"] = "Un bellissimo post per i social."
cap_res = agents.write_caption_node(state)
state.update(cap_res)
tag_res = agents.generate_hashtags_node(state)
db.update_draft(
draft.id,
title=cap_res.get("title", ""),
caption=cap_res.get("caption", ""),
hashtags=tag_res.get("hashtags", ""),
audio_analysis=analysis_res.get("audio_analysis", "")
)
st.rerun()
if st.button(f"✅ Salva Modifiche e Pubblica", key=f"pub_{draft.id}", type="primary", use_container_width=True):
db.update_draft(draft.id, title=edited_title, caption=edited_caption, hashtags=edited_tags)
full_text = f"*{edited_title}*\n\n{edited_caption}\n\n{edited_tags}"
# Recupera dati specifici artista (Token e Profile IDs)
artist_data = next((a for a in config['artists'] if a['id'] == draft.artist_id), None)
artist_token = artist_data.get("buffer_token") if artist_data else None
# Passiamo profile_ids vuoti per attivare l'auto-discovery nel publisher
profile_ids = []
res = publisher.publish(profile_ids, full_text, draft.image_path, draft.video_url, draft.image_paths, artist_token=artist_token)
if res['success']:
db.mark_as_published(draft.id)
st.success("Inviato a Buffer!")
st.rerun()
else:
st.error(f"Errore: {res['error']}")
if st.button("🗑️ Elimina Bozza", key=f"del_single_{draft.id}", type="secondary", use_container_width=True):
db.delete_draft(draft.id)
st.rerun()
# Sezione Sostituzione Media / Generazione Video Manuale
st.markdown("---")
st.write("📸 **Sostituisci Media / Genera Video (Opzionale)**")
st.caption("Seleziona una o più foto/video recenti per generare un video animato.")
recent_media = db.get_recent_media(draft.artist_id, limit=10)
selected_images = []
cols_img = st.columns(5)
for i, media in enumerate(recent_media):
with cols_img[i % 5]:
img_to_show = None
if os.path.exists(media.file_path):
img_to_show = media.file_path
elif media.source_url:
# MINIATURA DIRETTA DA GOOGLE DRIVE (Metodo più robusto)
img_to_show = f"https://drive.google.com/thumbnail?id={media.source_url}&sz=w400"
if img_to_show:
if media.file_type == 'video':
st.markdown(f"🎥 **Video**")
try:
# TENTATIVO CLOUD-FIRST (Lookup ID da database)
m_id = media.source_url
if not m_id:
m_rec = db.get_media_by_path(media.file_path)
if m_rec: m_id = m_rec.source_url
if m_id:
try:
import requests, base64
thumb_url = f"https://drive.google.com/thumbnail?id={m_id}&sz=w200"
response = requests.get(thumb_url, timeout=3)
if response.status_code == 200:
b64_img = base64.b64encode(response.content).decode()
st.markdown(f'<img src="data:image/jpeg;base64,{b64_img}" style="width:100%; border-radius: 5px;">', unsafe_allow_html=True)
else:
st.markdown(f'<img src="https://lh3.googleusercontent.com/u/0/d/{m_id}=w200" style="width:100%; border-radius: 5px;">', unsafe_allow_html=True)
except:
st.caption("🖼️ Cloud N/A")
else:
st.caption("🖼️ No ID")
except Exception:
st.caption("🖼️ Error")
if st.checkbox("Scegli", key=f"chk_m_{draft.id}_{media.id}", label_visibility="collapsed"):
selected_images.append(media.file_path)
if st.button("🎬 Genera Video Animato (CapCut style) con i media selezionati", key=f"gen_{draft.id}"):
if not selected_images:
st.error("Seleziona almeno un media!")
else:
with st.spinner("Generazione video in corso..."):
# Sincronizza/Scarica media
final_local_media = []
import requests
for img_p in selected_images:
if not os.path.exists(img_p):
m_rec = db.get_media_by_path(img_p)
if m_rec and m_rec.source_url:
try:
dl_url = f"https://drive.google.com/uc?id={m_rec.source_url}"
resp = requests.get(dl_url, timeout=15)
if resp.status_code == 200:
os.makedirs(os.path.dirname(img_p), exist_ok=True)
with open(img_p, 'wb') as f:
f.write(resp.content)
final_local_media.append(img_p)
except: pass
else:
final_local_media.append(img_p)
from video_generator import VideoGenerator
vg = VideoGenerator()
audio = db.get_audio_for_artist(draft.artist_id)
if not audio:
st.error("Nessun file audio trovato per questo artista!")
else:
audio_path = audio.file_path
if not os.path.exists(audio_path) and audio.source_url:
try:
dl_url = f"https://drive.google.com/uc?id={audio.source_url}"
resp = requests.get(dl_url, timeout=30)
if resp.status_code == 200:
os.makedirs(os.path.dirname(audio_path), exist_ok=True)
with open(audio_path, 'wb') as f:
f.write(resp.content)
except: pass
try:
# Percorso ASSOLUTO interno a Docker per garantire il sync con il volume
out_base = "/app/data/outputs"
os.makedirs(out_base, exist_ok=True)
safe_artist = draft.artist_id.replace(" ", "")
out_path = f"{out_base}/{safe_artist}_{draft.id}.mp4"
vid_path = vg.generate_video(final_local_media, audio_path, edited_title, start_time=0, output_path=out_path)
thumb_path = vg.generate_thumbnail(vid_path)
video_url = vg.upload_to_ephemeral(vid_path)
if not video_url:
st.warning("Upload remoto fallito. Il video è disponibile solo localmente.")
db.update_draft(draft.id, video_url=video_url, video_path=vid_path)
st.rerun()
except Exception as e:
st.error(f"Errore generazione video: {e}")
# ============================================================
# STATO CONSUMO CREDITI HIGGSFIELD AI (In basso alla console)
# ============================================================
st.markdown("---")
st.subheader("📊 Stato Consumo Crediti Higgsfield AI")
credits_data = load_credits_cache()
total = credits_data.get("total_monthly", 1000)
remaining = credits_data.get("remaining", 1000)
consumed = total - remaining
# Visualizzazione badge di stato del sync online
sync_status = credits_data.get("last_sync_status", "no_token")
sync_time = credits_data.get("last_sync_time", "Mai")
sync_err = credits_data.get("sync_error", "")
if sync_status == "success":
st.markdown(f"<span style='background-color:#d4ff00; color:#000000; padding:4px 8px; border-radius:4px; font-weight:bold; font-size:12px;'>🟢 SINCRONIZZATO ONLINE</span> &nbsp;&nbsp;<span style='font-size:12px; color:#888888;'>Ultimo aggiornamento automatico: {sync_time}</span>", unsafe_allow_html=True)
elif sync_status == "failed":
st.markdown(f"<span style='background-color:#ff4b4b; color:#ffffff; padding:4px 8px; border-radius:4px; font-weight:bold; font-size:12px;'>🔴 ERRORE DI CONNESSE</span> &nbsp;&nbsp;<span style='font-size:12px; color:#888888;'>Impossibile allineare i dati online ({sync_err}). Utilizzo cache del {sync_time}</span>", unsafe_allow_html=True)
else:
st.markdown("<span style='background-color:#ffaa00; color:#000000; padding:4px 8px; border-radius:4px; font-weight:bold; font-size:12px;'>🟡 CONTATORE LOCALE (CACHE)</span> &nbsp;&nbsp;<span style='font-size:12px; color:#888888;'>Token .env non configurato</span>", unsafe_allow_html=True)
st.info("💡 **Allineamento Online in Tempo Reale**: Per allineare automaticamente il consumo effettivo con il tuo account Higgsfield AI (Pro Plan) senza usare Chrome Safe Storage, aggiungi la variabile `HIGGSFIELD_TOKEN=eyJ...` nel file `.env` (copiabile dal cookie `__session` della console browser di higgsfield.ai).")
# Calculate percentage
pct_remaining = (remaining / total * 100) if total > 0 else 0.0
# Render elegant columns
col_c1, col_c2, col_c3 = st.columns(3)
col_c1.metric("Crediti Rimanenti", f"{remaining} crediti", f"{pct_remaining:.1f}% rimasti", delta_color="normal")
col_c2.metric("Crediti Consumati", f"{consumed} crediti", f"{100 - pct_remaining:.1f}% consumati", delta_color="inverse")
col_c3.metric("Budget Mensile", f"{total} crediti", help="Questo valore è allineato in tempo reale con il tuo abbonamento online")
# Beautiful, customized progress bar matching Higgsfield's neon lime styling
progress_html = f"""
<div style="font-weight: bold; font-size: 16px; margin-top: 15px; margin-bottom: 5px; font-family: sans-serif;">
{pct_remaining:.0f}% credits left
</div>
<div style="background-color: #333333; border-radius: 10px; width: 100%; height: 16px; overflow: hidden; margin-bottom: 25px; border: 1px solid #222222; box-shadow: inset 0 1px 3px rgba(0,0,0,0.2);">
<div style="background: linear-gradient(90deg, #d4ff00 0%, #bfff00 100%); width: {pct_remaining:.1f}%; height: 100%; transition: width 0.5s ease-in-out;">
</div>
</div>
"""
st.markdown(progress_html, unsafe_allow_html=True)
# ============================================================
# AUTO-REFRESH (Solo se ci sono task in corso)
# ============================================================
if running_tasks:
import time
time.sleep(5)
st.rerun()
+54
View File
@@ -0,0 +1,54 @@
import os
import google.generativeai as genai
from dotenv import load_dotenv
load_dotenv()
class AudioAnalyzer:
def __init__(self, api_key=None):
self.api_key = api_key or os.getenv("GOOGLE_API_KEY")
if self.api_key:
genai.configure(api_key=self.api_key)
self.model = genai.GenerativeModel('gemini-2.5-flash')
else:
self.model = None
def analyze_audio(self, file_path):
if not self.model:
return "API Key non configurata."
if not os.path.exists(file_path):
return f"File non trovato: {file_path}"
try:
# Rilevamento mime-type per evitare errori
mime_type = "audio/mpeg"
if file_path.endswith(".wav"): mime_type = "audio/wav"
elif file_path.endswith(".ogg"): mime_type = "audio/ogg"
# Upload file to Gemini con mime_type esplicito
print(f"🎵 Caricamento audio su Gemini: {file_path}...")
audio_file = genai.upload_file(path=file_path, mime_type=mime_type)
# Wait for processing
while audio_file.state.name == "PROCESSING":
import time
print("...elaborazione audio in corso su Gemini...")
time.sleep(2)
audio_file = genai.get_file(audio_file.name)
print("🧠 Analisi audio in corso con AI...")
prompt = """
Analizza questo brano musicale e rispondi in formato JSON con queste chiavi:
1. "climax_start_sec": (intero) Il secondo esatto in cui inizia la parte più energica o il ritornello (es. 45). ASSICURATI che in questo punto l'audio sia chiaramente udibile (evita silenzi iniziali o intro troppo lunghe).
2. "bpm": (intero) I battiti per minuto del brano (es. 128). Se non sei sicuro, fornisci una stima accurata basata sul ritmo.
3. "analysis": Scrivi un "Pitch" accattivante e professionale (max 3-4 frasi) pronto da inviare ai curatori delle playlist di Spotify, mettendo in luce il genere, il mood, le vibes e i punti di forza del brano.
Rispondi SOLO ed ESCLUSIVAMENTE con il JSON puro. Non includere blocchi markdown come ```json o ```, inizia direttamente con { e finisci con }.
"""
response = self.model.generate_content([prompt, audio_file])
return response.text
except Exception as e:
return f"Errore durante l'analisi audio: {str(e)}"
+146
View File
@@ -0,0 +1,146 @@
import os
import sys
import json
import time
import argparse
from database import Database
from agents import SocialAgents
def create_lock(task_name, artist_id="all"):
os.makedirs("data/tasks", exist_ok=True)
lock_file = f"data/tasks/{task_name}_{artist_id.replace(' ', '_')}.lock"
with open(lock_file, "w") as f:
f.write(str(os.getpid()))
return lock_file
def remove_lock(lock_file):
if os.path.exists(lock_file):
os.remove(lock_file)
def run_sync():
print("🚀 Avvio Sincronizzazione in background...")
import cloud_sync
cloud_sync.sync_from_drive()
print("✅ Sincronizzazione completata.")
def run_generate(artist_id, mode="both"):
db = Database()
agents = SocialAgents()
with open('config.json', 'r') as f:
config = json.load(f)
artist = next((a for a in config['artists'] if a['id'] == artist_id), None)
if not artist:
print(f"❌ Artista non trovato: {artist_id}")
return
print(f"🤖 Avvio generazione ({mode}) per {artist['name']}...")
# Sincronizzazione preventiva (leggera)
import cloud_sync
cloud_sync.sync_from_drive()
results = agents.run_for_artist(artist, mode=mode)
for res in results:
if 'error' not in res:
db.save_draft(
artist_id=artist_id,
title=res.get('title', 'Nuovo Post'),
caption=res.get('caption', ''),
hashtags=res.get('hashtags', ''),
image_path=res.get('image_path'),
image_url=res.get('image_url'),
video_url=res.get('video_url'),
audio_analysis=res.get('audio_analysis'),
video_path=res.get('video_path'),
focus_points=res.get('focus_points')
)
print(f"✅ Generazione completata per {artist['name']}.")
def run_higgsfield(artist_id, prompt, dry_run=False):
print(f"🎨 Avvio task Higgsfield per {artist_id}...")
with open('config.json', 'r') as f:
config = json.load(f)
artist = next((a for a in config['artists'] if a['id'] == artist_id), None)
if not artist:
print(f"❌ Artista non trovato: {artist_id}")
return
models_url = artist.get('drive_starting_photos_url')
if not models_url:
print(f"❌ Nessuna cartella Drive Foto Partenza configurata per {artist_id}")
return
soul_id = artist.get('soul_id')
if not soul_id:
print(f"❌ Nessun Higgsfield Soul ID configurato per {artist_id}")
return
# 1. Download starting photos from Drive
import cloud_sync
import gdown
fid = cloud_sync.get_drive_id(models_url)
local_dir = f"data/starting_photos/{artist_id.replace(' ', '_')}"
os.makedirs(local_dir, exist_ok=True)
print(f"📥 Controllo cartella Drive per modelli/foto partenza ({fid})...")
files = cloud_sync.list_files_in_public_folder(fid)
if not files:
print(f"⚠️ Nessun file trovato nella cartella Drive: {models_url}")
for fid_sub, fname in files:
if fname.lower().endswith(('.jpg', '.jpeg', '.png', '.webp')):
local_path = f"{local_dir}/{fname}"
if not os.path.exists(local_path):
print(f"📥 Scaricamento foto partenza: {fname}...")
url = f"https://drive.google.com/uc?id={fid_sub}"
try:
gdown.download(url, output=local_path, quiet=True)
except Exception as dl_err:
print(f"⚠️ Errore download {fname}: {dl_err}")
# 2. Run generate_higgsfield_photos.py via subprocess
cmd = [
sys.executable, "-m", "src.generate_higgsfield_photos",
"--dir", local_dir,
"--prompt", prompt
]
if dry_run:
cmd.append("--dry-run")
print(f"🚀 Esecuzione script generatore: {' '.join(cmd)}")
res = subprocess.run(cmd, capture_output=True, text=True)
print(res.stdout)
if res.stderr:
print(f"⚠️ Stderr: {res.stderr}")
print("✅ Task Higgsfield completato.")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--task", choices=["sync", "generate", "higgsfield"], required=True)
parser.add_argument("--artist_id", default="all")
parser.add_argument("--mode", choices=["photo", "video", "both"], default="both")
parser.add_argument("--prompt", default="A beautiful editorial studio portrait, highly detailed, cinematic studio lighting, professional photography, 8k resolution, crisp details")
parser.add_argument("--dry_run", action="store_true")
args = parser.parse_args()
lock = create_lock(args.task, f"{args.artist_id}_{args.mode}")
try:
if args.task == "sync":
run_sync()
elif args.task == "generate":
run_generate(args.artist_id, mode=args.mode)
elif args.task == "higgsfield":
run_higgsfield(args.artist_id, args.prompt, dry_run=args.dry_run)
except Exception as e:
import traceback
error_msg = f"❌ Errore critico nel task {args.task} ({args.artist_id}): {str(e)}\n{traceback.format_exc()}"
print(error_msg)
os.makedirs("data/tasks", exist_ok=True)
with open("data/tasks/error.log", "a") as f:
f.write(f"\n--- {time.ctime()} ---\n{error_msg}\n")
finally:
remove_lock(lock)
+153
View File
@@ -0,0 +1,153 @@
import os
import requests
import json
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
class BufferPublisher:
def __init__(self, token=None):
self.token = token or os.getenv("BUFFER_ACCESS_TOKEN")
self.graphql_url = "https://api.buffer.com/graphql"
self.log_file = "data/buffer_logs.txt"
def log(self, message):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
os.makedirs("data", exist_ok=True)
with open(self.log_file, "a") as f:
f.write(f"[{timestamp}] {message}\n")
def get_profiles(self):
query = """
query {
profiles {
id
type
service
}
}
"""
headers = {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json"
}
try:
response = requests.post(self.graphql_url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
profiles = data.get('data', {}).get('profiles', [])
# Filtriamo per i servizi che ci interessano (Instagram e TikTok)
return [p['id'] for p in profiles if p['service'] in ['instagram', 'tiktok']]
else:
self.log(f"ERRORE recupero profili: {response.text}")
return []
except Exception as e:
self.log(f"ECCEZIONE recupero profili: {e}")
return []
def publish(self, profile_ids, text, image_path, video_url=None, image_paths=None, artist_token=None):
# Garantisce l'uso del token specifico o il fallback al globale per ogni singola chiamata
self.token = artist_token or os.getenv("BUFFER_ACCESS_TOKEN")
# Se non ci sono profile_ids, li cerchiamo automaticamente
if not profile_ids:
self.log("Nessun Profile ID fornito, avvio auto-discovery...")
profile_ids = self.get_profiles()
if not profile_ids:
return {"success": False, "error": "Nessun canale social trovato per questo account Buffer."}
self.log(f"Canali trovati automaticamente: {len(profile_ids)}")
return self.publish_via_graphql(profile_ids, text, image_path, video_url, image_paths)
def publish_via_graphql(self, profile_ids, text, image_path, video_url=None, image_paths=None):
from database import Database, Media
db = Database()
session = db.Session()
# Prepariamo gli asset: se c'è un video, ha la precedenza
assets_payload = None
if video_url:
assets_payload = {"videos": [{"url": video_url}]}
elif image_paths:
# Gestione Carousel (lista di percorsi locali)
if isinstance(image_paths, str):
try: image_paths = json.loads(image_paths)
except: image_paths = [image_paths]
image_urls = []
for path in image_paths:
media = session.query(Media).filter_by(file_path=path).first()
if media and media.source_url:
image_urls.append({"url": f"https://lh3.googleusercontent.com/d/{media.source_url}=s1080"})
if image_urls:
assets_payload = {"images": image_urls}
else:
media = session.query(Media).filter_by(file_path=image_path).first()
if media and media.source_url:
direct_link = f"https://lh3.googleusercontent.com/d/{media.source_url}=s1080"
assets_payload = {"images": [{"url": direct_link}]}
results = []
for channel_id in profile_ids:
mutation = """
mutation ($input: CreatePostInput!) {
createPost(input: $input) {
... on PostActionSuccess { post { id } }
... on MutationError { message }
}
}
"""
# Input base per la PUBBLICAZIONE DIRETTA
post_input = {
"channelId": channel_id,
"text": text,
"schedulingType": "automatic", # Questo campo è obbligatorio per l'API anche se condividiamo subito
"mode": "shareNow" # Ignora la coda e pubblica ISTANTANEAMENTE
}
if assets_payload:
post_input["assets"] = assets_payload
metadata = {}
if channel_id == "69f9ea855c4c051afa117baa": # Instagram
metadata["instagram"] = {
"type": "post",
"shouldShareToFeed": True # CAMPO MANCANTE RIPRISTINATO
}
elif channel_id == "69f9e7e35c4c051afa116a9e": # TikTok
metadata["tiktok"] = {"title": text[:50]}
if metadata:
post_input["metadata"] = metadata
headers = {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json"
}
self.log(f"INVIO POST a Canale: {channel_id}")
response = requests.post(self.graphql_url, headers=headers, json={
'query': mutation,
'variables': {"input": post_input}
})
if response.status_code == 200:
data = response.json()
res = data.get('data', {}).get('createPost', {})
self.log(f"RISPOSTA BUFFER: {json.dumps(res)}")
if 'message' in res:
results.append({"success": False, "error": res['message']})
else:
results.append({"success": True, "id": res.get('post', {}).get('id')})
else:
self.log(f"ERRORE HTTP {response.status_code}: {response.text}")
results.append({"success": False, "error": response.text})
session.close()
if any(r['success'] for r in results):
return {"success": True, "details": results}
return {"success": False, "error": results[0].get('error', 'Errore')}
+120
View File
@@ -0,0 +1,120 @@
import os
import json
import requests
import re
import subprocess
from database import Database
db = Database()
def get_drive_id(url):
match = re.search(r'/file/d/([a-zA-Z0-9_-]+)', url)
if match: return match.group(1)
match = re.search(r'id=([a-zA-Z0-9_-]+)', url)
if match: return match.group(1)
match = re.search(r'folders/([a-zA-Z0-9_-]+)', url)
if match: return match.group(1)
return None
def list_files_in_public_folder(folder_id):
url = f"https://drive.google.com/embeddedfolderview?id={folder_id}"
files_found = []
try:
response = requests.get(url, timeout=15)
if response.status_code == 200:
entries = re.findall(r'id="entry-([a-zA-Z0-9_-]+)".*?class="flip-entry-title">([^<]+)</div>', response.text, re.DOTALL)
for fid, fname in entries:
if fid not in [f[0] for f in files_found]:
files_found.append((fid, fname))
except Exception as e:
print(f"Errore Drive: {e}")
return files_found
def generate_video_thumbnail(video_fid, output_path):
"""Genera una miniatura da un video di Google Drive usando FFmpeg"""
if os.path.exists(output_path):
print(f"⏩ Miniatura già esistente: {output_path}")
return True
print(f"🎬 Generazione miniatura per FID: {video_fid}...")
# URL di download diretto
url = f"https://drive.google.com/uc?id={video_fid}&export=download"
# Comando FFmpeg per estrarre un frame a 1 secondo
# Usiamo parametri per velocizzare l'apertura dello stream
cmd = [
'ffmpeg', '-ss', '00:00:01', '-i', url,
'-frames:v', '1', '-q:v', '2',
'-vf', 'scale=320:-1',
output_path, '-y'
]
try:
# Timeout per evitare blocchi infiniti su file troppo grandi o link protetti
res = subprocess.run(cmd, capture_output=True, timeout=25)
if res.returncode != 0:
print(f"⚠️ FFmpeg ha restituito un errore (probabile file protetto o troppo grande): {video_fid}")
return False
return os.path.exists(output_path)
except subprocess.TimeoutExpired:
print(f"⏳ Timeout FFmpeg per {video_fid} - saltato.")
return False
except Exception as e:
print(f"❌ Errore inaspettato thumbnail: {e}")
return False
def sync_from_drive():
if not os.path.exists('config.json'): return
# Cartella cache unica
cache_dir = "data/cache"
os.makedirs(cache_dir, exist_ok=True)
with open('config.json', 'r') as f:
config = json.load(f)
for artist in config['artists']:
artist_id = artist['id']
# Audio Discovery
audio_url = artist.get('drive_audio_url')
if audio_url:
fid = get_drive_id(audio_url)
if "/file/d/" in audio_url:
db.add_media(f"{cache_dir}/audio_{fid}", 'audio', artist_id, source_url=fid)
elif fid:
for fid_sub, fname in list_files_in_public_folder(fid):
db.add_media(f"{cache_dir}/{fname}", 'audio', artist_id, source_url=fid_sub)
# Images Discovery
images_url = artist.get('drive_images_url')
if images_url:
fid = get_drive_id(images_url)
if "/file/d/" in images_url:
db.add_media(f"{cache_dir}/image_{fid}", 'image', artist_id, source_url=fid)
elif fid:
for fid_sub, fname in list_files_in_public_folder(fid):
db.add_media(f"{cache_dir}/{fname}", 'image', artist_id, source_url=fid_sub)
# Videos Discovery
videos_url = artist.get('drive_videos_url')
if videos_url:
fid = get_drive_id(videos_url)
if not fid:
print(f"⚠️ URL Video non valido per {artist_id}: {videos_url}")
elif "/file/d/" in videos_url:
# Nessuna miniatura locale per risparmiare spazio
db.add_media(f"{cache_dir}/video_{fid}", 'video', artist_id, source_url=fid)
else:
v_files = list_files_in_public_folder(fid)
if not v_files:
print(f"️ Nessun video trovato nella cartella di {artist_id}")
for fid_sub, fname in v_files:
# Registriamo solo il metadata, la miniatura sarà caricata 'on-the-fly' dalla UI
db.add_media(f"{cache_dir}/{fname}", 'video', artist_id, source_url=fid_sub)
else:
print(f"️ Nessuna cartella video configurata per {artist_id}")
if __name__ == "__main__":
sync_from_drive()
print("Sincro Cache completata.")
+239
View File
@@ -0,0 +1,239 @@
import datetime
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class Media(Base):
__tablename__ = 'media'
id = Column(Integer, primary_key=True)
file_path = Column(String, unique=True, nullable=False)
file_type = Column(String) # 'image' or 'audio'
artist_id = Column(String)
is_used = Column(Boolean, default=False)
description = Column(Text)
source_url = Column(String)
thumbnail_path = Column(String) # NUOVO
created_at = Column(DateTime, default=datetime.datetime.utcnow)
last_used_at = Column(DateTime)
class Draft(Base):
__tablename__ = 'drafts'
id = Column(Integer, primary_key=True)
artist_id = Column(String)
title = Column(String)
caption = Column(Text)
hashtags = Column(Text)
image_path = Column(String)
image_paths = Column(Text) # JSON list of local paths
image_url = Column(String) # NUOVA COLONNA PER ANTEPRIMA DIRETTA
video_url = Column(String)
video_path = Column(String) # NUOVO
audio_analysis = Column(Text)
focus_points = Column(Text) # NUOVO: Coordinate JSON per la regia AI
status = Column(String, default='pending') # pending, approved, published
created_at = Column(DateTime, default=datetime.datetime.utcnow)
class Database:
def __init__(self, db_url="sqlite:///data/redazione.db"):
self.engine = create_engine(db_url)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
def add_media(self, file_path, file_type, artist_id, description=None, source_url=None, thumbnail_path=None):
session = self.Session()
try:
media = session.query(Media).filter_by(file_path=file_path).first()
if not media:
media = Media(
file_path=file_path,
file_type=file_type,
artist_id=artist_id,
description=description,
source_url=source_url,
thumbnail_path=thumbnail_path
)
session.add(media)
session.commit()
elif thumbnail_path and not media.thumbnail_path:
media.thumbnail_path = thumbnail_path
session.commit()
return media
finally:
session.close()
def get_unused_image(self, artist_id):
from sqlalchemy.sql.expression import func
session = self.Session()
try:
return session.query(Media).filter_by(
artist_id=artist_id,
file_type='image',
is_used=False
).order_by(func.random()).first()
finally:
session.close()
def get_media_by_artist(self, artist_id, file_type):
session = self.Session()
try:
return session.query(Media).filter_by(
artist_id=artist_id,
file_type=file_type
).first()
finally:
session.close()
def get_audio_for_artist(self, artist_id):
return self.get_media_by_artist(artist_id, 'audio')
def mark_as_used(self, media_id):
session = self.Session()
try:
media = session.query(Media).get(media_id)
if media:
media.is_used = True
media.last_used_at = datetime.datetime.utcnow()
session.commit()
finally:
session.close()
def get_audio_for_artist(self, artist_id):
# We assume one main audio track per session or the latest added
session = self.Session()
try:
return session.query(Media).filter_by(
artist_id=artist_id,
file_type='audio'
).order_by(Media.created_at.desc()).first()
finally:
session.close()
def save_draft(self, artist_id, title, caption, hashtags, image_path, audio_analysis, image_url=None, video_url=None, image_paths=None, video_path=None, focus_points=None):
session = self.Session()
try:
draft = Draft(
artist_id=artist_id,
title=title,
caption=caption,
hashtags=hashtags,
image_path=image_path,
image_paths=image_paths,
image_url=image_url,
video_url=video_url,
video_path=video_path,
audio_analysis=audio_analysis,
focus_points=focus_points # SALVATAGGIO
)
session.add(draft)
session.commit()
return draft.id
finally:
session.close()
def get_recent_media(self, artist_id, limit=10):
session = self.Session()
try:
return session.query(Media).filter(
Media.artist_id == artist_id,
Media.file_type.in_(['image', 'video'])
).order_by(Media.created_at.desc()).limit(limit).all()
finally:
session.close()
def get_mixed_assets(self, artist_id, total=12):
from sqlalchemy.sql.expression import func
session = self.Session()
try:
# 1. Seleziona una copertina (preferibilmente immagine mai usata)
cover = session.query(Media).filter_by(
artist_id=artist_id,
file_type='image',
is_used=False
).order_by(func.random()).first()
if not cover:
cover = session.query(Media).filter_by(
artist_id=artist_id,
file_type='image'
).order_by(func.random()).first()
if not cover:
# Se proprio non ci sono immagini, prendi un video come cover
cover = session.query(Media).filter_by(
artist_id=artist_id,
file_type='video'
).order_by(func.random()).first()
if not cover:
return []
# 2. Seleziona un mix di altre immagini e video
others = session.query(Media).filter(
Media.artist_id == artist_id,
Media.file_type.in_(['image', 'video']),
Media.id != cover.id
).order_by(func.random()).limit(total - 1).all()
return [cover] + others
finally:
session.close()
def get_pending_drafts(self, artist_id=None):
session = self.Session()
try:
query = session.query(Draft).filter_by(status='pending')
if artist_id:
query = query.filter_by(artist_id=artist_id)
return query.all()
finally:
session.close()
def get_media_by_path(self, file_path):
session = self.Session()
try:
return session.query(Media).filter_by(file_path=file_path).first()
finally:
session.close()
def delete_draft(self, draft_id):
session = self.Session()
try:
draft = session.query(Draft).filter_by(id=draft_id).first()
if draft:
session.delete(draft)
session.commit()
finally:
session.close()
def update_draft(self, draft_id, title=None, caption=None, hashtags=None, image_path=None, status=None, audio_analysis=None, video_url=None, video_path=None, image_paths=None, focus_points=None):
session = self.Session()
try:
draft = session.query(Draft).get(draft_id)
if draft:
if title is not None: draft.title = title
if caption is not None: draft.caption = caption
if hashtags is not None: draft.hashtags = hashtags
if image_path is not None: draft.image_path = image_path
if status is not None: draft.status = status
if audio_analysis is not None: draft.audio_analysis = audio_analysis
if video_url is not None: draft.video_url = video_url
if video_path is not None: draft.video_path = video_path
if image_paths is not None: draft.image_paths = image_paths
if focus_points is not None: draft.focus_points = focus_points # AGGIORNAMENTO
session.commit()
return True
return False
finally:
session.close()
def mark_as_published(self, draft_id):
session = self.Session()
try:
draft = session.query(Draft).filter_by(id=draft_id).first()
if draft:
draft.status = 'published'
session.commit()
finally:
session.close()
+86
View File
@@ -0,0 +1,86 @@
import os
import json
import sys
from dotenv import load_dotenv
# Aggiungiamo src al path
sys.path.append(os.path.join(os.getcwd(), 'src'))
from database import Database
from agents import SocialAgents
from buffer_publisher import BufferPublisher
load_dotenv()
def force_test():
db = Database()
agents = SocialAgents()
publisher = BufferPublisher()
# 1. Carichiamo i dati di Veronica
with open('config.json', 'r') as f:
config = json.load(f)
veronica = next(a for a in config['artists'] if "Veronica" in a['name'])
print(f"--- FASE 1: GENERAZIONE PER {veronica['name']} ---")
result = agents.run_for_artist(veronica)
if 'error' in result:
print(f"Errore Generazione: {result['error']}")
return
print(f"Post Generato: {result['title']}")
# 2. Salvataggio bozza locale
draft_id = db.save_draft(
artist_id=veronica['id'],
title=result['title'],
caption=result['caption'],
hashtags=result['hashtags'],
image_path=result['image_path'],
audio_analysis=result['audio_analysis']
)
# 3. PUBBLICAZIONE SU BUFFER
print(f"\n--- FASE 2: INVIO A BUFFER ---")
profile_ids = ["69f9e7e35c4c051afa116a9e", "69f9ea855c4c051afa117baa"]
full_text = f"*{result['title']}*\n\n{result['caption']}\n\n{result['hashtags']}"
pub_res = publisher.publish(profile_ids, full_text, result['image_path'])
print(f"Risultato Pubblicazione: {pub_res}")
# 4. VERIFICA REALE SU BUFFER
print(f"\n--- FASE 3: VERIFICA BOZZE SU BUFFER ---")
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
# Chiediamo i post in stato 'draft' per Instagram
query = """
query {
channels(input: {ids: ["69f9ea855c4c051afa117baa"]}) {
name
posts(input: {state: draft}) {
totalCount
nodes {
id
text
}
}
}
}
"""
resp = requests.post(url, headers=headers, json={'query': query})
if resp.status_code == 200:
data = resp.json()
drafts = data['data']['channels'][0]['posts']
print(f"Bozze totali su Instagram: {drafts['totalCount']}")
for node in drafts['nodes']:
print(f"- Bozza trovata ID: {node['id']} | Testo: {node['text'][:50]}...")
else:
print(f"Errore verifica: {resp.text}")
if __name__ == "__main__":
import requests # Lo importiamo qui per sicurezza
force_test()
+401
View File
@@ -0,0 +1,401 @@
import os
import sys
import json
import argparse
import subprocess
import re
import requests
from pathlib import Path
from dotenv import load_dotenv
# Add the directory containing this script to sys.path to allow imports from src
current_dir = Path(__file__).resolve().parent
if str(current_dir) not in sys.path:
sys.path.insert(0, str(current_dir))
# Load environment variables
load_dotenv()
def normalize_string(s):
"""Normalize a string to lowercase and remove non-alphanumeric characters for fuzzy matching."""
if not s:
return ""
return re.sub(r'[^a-z0-9]', '', s.lower())
def load_artists_config(config_path="config.json"):
"""Load artists configuration from config.json."""
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
return config.get("artists", [])
except Exception as e:
print(f"❌ Error loading config.json: {e}", file=sys.stderr)
return []
def match_artist(file_path, artists):
"""
Fuzzy match a file path/filename to an artist in config.json.
Returns the matched artist dict, or None if no match found.
"""
path_obj = Path(file_path)
# Check parent directory name and filename
search_space = f"{path_obj.parent.name} {path_obj.name}"
normalized_search = normalize_string(search_space)
# Try exact or substring matches
for artist in artists:
artist_id = artist.get("id", "")
artist_name = artist.get("name", "")
social_tag = artist.get("social_tag", "")
norm_id = normalize_string(artist_id)
norm_name = normalize_string(artist_name)
norm_tag = normalize_string(social_tag)
# Check if artist name/id is in the search space
if (norm_id and norm_id in normalized_search) or \
(norm_name and norm_name in normalized_search) or \
(norm_tag and norm_tag in normalized_search):
return artist
# Fuzzy match check (e.g. if singer first name matches)
for artist in artists:
artist_name = artist.get("name", "")
first_name = artist_name.split()[0] if artist_name else ""
norm_first = normalize_string(first_name)
if norm_first and norm_first in normalized_search:
return artist
return None
def run_higgsfield_cli_generate(prompt, soul_id, image_path, output_path):
"""
Run Higgsfield generation using the Higgsfield CLI via subprocess.
"""
print(f"🎬 [CLI] Generating AI photo for Soul ID: {soul_id}...")
try:
# First, upload the reference image if provided
upload_cmd = ["higgsfield", "upload", str(image_path)]
print(f" Uploading reference image: {' '.join(upload_cmd)}")
upload_res = subprocess.run(upload_cmd, capture_output=True, text=True, check=True)
# Extract UUID from upload response
# Standard UUID regex
uuids = re.findall(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', upload_res.stdout, re.IGNORECASE)
if not uuids:
print(f" ❌ No UUID found in upload output: {upload_res.stdout}")
return False
uuid = uuids[0]
print(f" Uploaded successfully. Reference UUID: {uuid}")
# Create generation job
gen_cmd = [
"higgsfield", "generate", "create", "soul_v2",
"--prompt", prompt,
"--soul-id", soul_id,
"--image-id", uuid,
"--wait"
]
print(f" Running generation: {' '.join(gen_cmd)}")
gen_res = subprocess.run(gen_cmd, capture_output=True, text=True, check=True)
# Find output URL or download path
# If the CLI downloads automatically or provides a URL, we capture it
urls = re.findall(r'https?://[^\s]+', gen_res.stdout)
if urls:
url = urls[0]
print(f" Generation completed. Output URL: {url}")
# Download file
resp = requests.get(url, timeout=30)
if resp.status_code == 200:
with open(output_path, "wb") as f:
f.write(resp.content)
print(f" ✅ Saved generated photo to: {output_path}")
return True
else:
print(f" ❌ Could not extract output URL from generation log: {gen_res.stdout}")
return False
except subprocess.CalledProcessError as e:
print(f" ❌ CLI command failed: {e.stderr}")
return False
except Exception as e:
print(f" ❌ CLI generation error: {e}")
return False
def run_higgsfield_api_generate(prompt, model_to_use, soul_id, image_path, output_path, api_key, api_secret=None):
"""
Run Higgsfield generation using direct REST API requests via the official SDK.
"""
import higgsfield_client
if soul_id:
print(f"🔌 [API] Generating AI photo using model '{model_to_use}' for Soul ID: {soul_id}...")
else:
print(f"🔌 [API] Generating AI photo using fallback model '{model_to_use}'...")
try:
# Set environment variables for the SDK
os.environ["HF_API_KEY"] = api_key
if api_secret:
os.environ["HF_API_SECRET"] = api_secret
print(" Initializing Higgsfield SDK Client...")
client = higgsfield_client.SyncClient()
# 1. Upload starting photo
print(f" Uploading starting photo: {image_path}...")
public_url = client.upload_file(image_path)
print(f" Uploaded successfully. URL: {public_url}")
# 2. Submit Generation Job
if model_to_use == "flux-2":
arguments = {
"prompt": prompt,
"aspect_ratio": "1:1",
"resolution": "2k",
"input_images": [public_url]
}
else:
# Fallback/Default for soul or others if soul_id is present
# We use the 'soul' endpoint on the platform API
arguments = {
"prompt": prompt,
"aspect_ratio": "1:1",
"style_id": "realistic", # default style_id required by the API
"character_id": soul_id,
"input_images": [public_url]
}
print(f" Submitting job to endpoint '{model_to_use}'...")
controller = client.submit(
application=model_to_use,
arguments=arguments
)
request_id = controller.request_id
print(f" Job submitted successfully. Request ID: {request_id}. Polling for completion...")
# 3. Poll for completion
for status in controller.poll_request_status(delay=3.0):
# Print status to stdout for log visibility
print(f" Job status: {status}")
# 4. Download result
print(" Retrieving completed job data...")
result = controller.get()
# In flux-2/soul, completed output is stored in 'images' list or 'outputs' list
images = result.get("images") or result.get("outputs") or []
output_url = result.get("output_url") or (images[0].get("url") if images else None)
if not output_url:
print(f" ❌ Job completed but no output URL found in response: {result}")
return False
print(f" ✅ Generation complete! Downloading from: {output_url}")
resp = requests.get(output_url, timeout=30)
if resp.status_code == 200:
with open(output_path, "wb") as f:
f.write(resp.content)
print(f" ✅ Saved generated photo to: {output_path}")
return True
else:
print(f" ❌ Download failed with status: {resp.status_code}")
return False
except Exception as e:
print(f" ❌ REST API generation error: {e}")
import traceback
traceback.print_exc()
return False
def deduct_credits(amount=1):
try:
cache_dir = Path("data/cache")
cache_dir.mkdir(parents=True, exist_ok=True)
credits_file = cache_dir / "credits.json"
# Load existing or create
if credits_file.exists():
with open(credits_file, 'r') as f:
data = json.load(f)
else:
data = {"total_monthly": 1000, "remaining": 1000, "generated_this_month": 0}
data["remaining"] = max(0, data.get("remaining", 1000) - amount)
data["generated_this_month"] = data.get("generated_this_month", 0) + amount
with open(credits_file, 'w') as f:
json.dump(data, f, indent=2)
print(f" 📊 Local credits updated: {data['remaining']}/{data['total_monthly']} remaining ({data['generated_this_month']} generated)")
except Exception as e:
print(f" ⚠️ Could not update credits cache: {e}")
def main():
parser = argparse.ArgumentParser(description="Batch generate AI photos using Higgsfield AI with Soul ID character consistency.")
parser.add_argument("--dir", required=True, help="Directory containing starting photos.")
parser.add_argument("--soul-ids", help="JSON string mapping artist names/IDs to Higgsfield Soul IDs.")
parser.add_argument("--output-dir", default="data/images/generated_photos", help="Output directory.")
parser.add_argument("--prompt", default="A beautiful editorial studio portrait, highly detailed, cinematic studio lighting, professional photography, 8k resolution, crisp details", help="Prompt override for visual style.")
parser.add_argument("--dry-run", action="store_true", help="Simulate the execution and check mappings without making API/CLI requests.")
args = parser.parse_args()
input_dir = Path(args.dir)
if not input_dir.is_dir():
print(f"❌ Error: {input_dir} is not a valid directory.", file=sys.stderr)
sys.exit(1)
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# Load artists from config
artists = load_artists_config()
if not artists:
print("⚠️ Warning: No artists found in config.json.", file=sys.stderr)
# Load custom soul IDs mapping
custom_soul_ids = {}
if args.soul_ids:
try:
custom_soul_ids = json.loads(args.soul_ids)
print(f"️ Loaded custom Soul ID overrides: {custom_soul_ids}")
except Exception as e:
print(f"❌ Error parsing --soul-ids JSON: {e}", file=sys.stderr)
sys.exit(1)
# Compile final artist -> soul_id mapping
artist_soul_map = {}
for artist in artists:
name = artist.get("name", "")
artist_id = artist.get("id", "")
# Priority: 1. CLI Override by Name, 2. CLI Override by ID, 3. config.json soul_id
soul_id = custom_soul_ids.get(name) or custom_soul_ids.get(artist_id) or artist.get("soul_id")
if soul_id:
artist_soul_map[name] = soul_id
print(f"️ Active Singer Soul ID mapping: {artist_soul_map}")
# Scan input directory for images
supported_extensions = ['.png', '.jpg', '.jpeg', '.webp']
image_files = [f for f in input_dir.rglob("*") if f.suffix.lower() in supported_extensions]
if not image_files:
print(f"⚠️ No starting photos found in {input_dir}.")
sys.exit(0)
print(f"🔍 Found {len(image_files)} starting photos. Beginning processing...")
# Retrieve credentials
api_key = os.getenv("HIGGSFIELD_API_KEY")
api_secret = os.getenv("HIGGSFIELD_SECRET") # optional
# Determine generation method
use_api = bool(api_key and api_key != "your_higgsfield_api_key_here")
# If explicitly in dry-run mode
if args.dry_run:
print("\n🧪 [DRY-RUN] Simulating Higgsfield AI generation pipeline. No real API calls will be made.")
elif not use_api:
print("️ HIGGSFIELD_API_KEY is not set or is placeholder. Using CLI-based generation.")
# Check if higgsfield CLI is installed
try:
res = subprocess.run(["which", "higgsfield"], capture_output=True, text=True)
if res.returncode != 0:
print("⚠️ Warning: Higgsfield CLI ('higgsfield') not found on system path.")
print(" Automatically falling back to DRY-RUN simulation mode.")
args.dry_run = True
except Exception as e:
print("⚠️ Warning: Error checking for Higgsfield CLI.")
print(" Automatically falling back to DRY-RUN simulation mode.")
args.dry_run = True
success_count = 0
failure_count = 0
for img_path in image_files:
print(f"\n📸 Processing photo: {img_path.name}")
# 1. Match artist
matched_artist = match_artist(img_path, artists)
if not matched_artist:
print(f" ⚠️ Skipping: Could not match image path/name to any known artist in config.json.")
failure_count += 1
continue
artist_name = matched_artist.get("name")
print(f" Matched Artist: {artist_name}")
# 2. Retrieve Soul ID & Check if valid
soul_id = artist_soul_map.get(artist_name)
is_placeholder = bool(soul_id and (soul_id.startswith("soul_") or soul_id == "your_soul_id_here"))
if not soul_id or is_placeholder:
print(f" ️ No valid Soul ID configured for artist '{artist_name}'. Falling back to flux-2 model!")
model_to_use = "flux-2"
soul_id = None
else:
model_to_use = "soul"
# 3. Generate output file name
out_filename = f"gen_{artist_name.replace(' ', '_')}_{img_path.stem}.png"
out_path = output_dir / out_filename
# 4. Generate AI Photo
if args.dry_run:
print(f" 🧪 [DRY-RUN] Would generate AI photo:")
print(f" - Starting photo: {img_path.absolute()}")
print(f" - Model to use: {model_to_use}")
print(f" - Soul ID reference: {soul_id}")
print(f" - Style prompt: {args.prompt}")
print(f" - Output destination: {out_path.absolute()}")
success = True
# Simulate DB save in dry-run
print(f" 🧪 [DRY-RUN] Would register in webapp database as Draft for {artist_name}")
elif use_api:
success = run_higgsfield_api_generate(args.prompt, model_to_use, soul_id, img_path, out_path, api_key, api_secret)
else:
# CLI fallback doesn't support model argument in our signature, so use original
success = run_higgsfield_cli_generate(args.prompt, soul_id, img_path, out_path)
if success:
success_count += 1
if not args.dry_run:
try:
from database import Database
db_instance = Database()
if model_to_use == "flux-2":
analysis_text = "Immagine generata tramite Higgsfield AI utilizzando il modello Flux-2 con riferimento visivo."
hashtags_text = "#higgsfield #flux #consistentcharacter"
else:
analysis_text = f"Immagine generata tramite Higgsfield AI utilizzando il Soul ID: {soul_id}."
hashtags_text = "#higgsfield #soul #consistentcharacter"
draft_id = db_instance.save_draft(
artist_id=matched_artist.get("id"),
title=f"AI Photo - {artist_name}",
caption=f"Ecco un nuovo post generato con l'intelligenza artificiale per {artist_name}! #music #ai",
hashtags=hashtags_text,
image_path=str(out_path),
audio_analysis=analysis_text
)
print(f" 💾 Registered in webapp database as Draft ID: {draft_id}")
deduct_credits(1)
except Exception as db_err:
print(f" ⚠️ Could not register in webapp database: {db_err}")
else:
failure_count += 1
print(f"\n📊 Processing complete!")
print(f" Successfully processed/simulated: {success_count} photos")
print(f" Failed or skipped: {failure_count} photos")
if args.dry_run:
print(f" 🧪 Dry-run simulation completed successfully. No files were written to: {output_dir.absolute()}")
else:
print(f" Generated photos are saved in: {output_dir.absolute()}")
if __name__ == "__main__":
main()
+35
View File
@@ -0,0 +1,35 @@
import json
import asyncio
from agents import SocialAgents
from database import Database
async def generate():
db = Database()
agents = SocialAgents()
with open('config.json', 'r') as f:
config = json.load(f)
artists = config['artists']
for artist in artists:
try:
print(f"Generazione per {artist['name']}...")
result = agents.run_for_artist(artist)
if result['caption']:
db.save_draft(
artist_id=artist['id'],
title=result['title'],
caption=result['caption'],
hashtags=result['hashtags'],
image_path=result['image_path'],
audio_analysis=result['audio_analysis']
)
print(f"✅ Bozza salvata per {artist['name']}")
else:
print(f"⚠️ Nessun contenuto generato per {artist['name']} (controlla se ci sono file)")
except Exception as e:
print(f"❌ Errore per {artist['name']}: {str(e)}")
if __name__ == "__main__":
asyncio.run(generate())
+36
View File
@@ -0,0 +1,36 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def inspect_asset_detail():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
query = """
query {
__type(name: "ImageAssetInput") {
inputFields {
name
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
fields = data['data']['__type']['inputFields']
print("\n--- CAMPI DI IMAGEASSETINPUT ---")
for f in fields:
print(f"Campo: {f['name']}")
else:
print(f"Errore {response.status_code}")
if __name__ == "__main__":
inspect_asset_detail()
+28
View File
@@ -0,0 +1,28 @@
import os
import requests
import json
from dotenv import load_dotenv
load_dotenv()
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
query = """
query {
__type(name: "AssetsInput") {
inputFields {
name
type {
name
kind
ofType { name kind }
}
}
}
}
"""
resp = requests.post(url, headers=headers, json={'query': query})
print(json.dumps(resp.json(), indent=2))
+45
View File
@@ -0,0 +1,45 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def inspect_metadata():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Cerchiamo il tipo del campo 'metadata'
query = """
query {
__type(name: "CreatePostInput") {
inputFields {
name
type {
name
kind
ofType {
name
kind
}
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
fields = data['data']['__type']['inputFields']
for f in fields:
if f['name'] == 'metadata':
print(f"Tipo di metadata: {f['type']['name'] or f['type']['ofType']['name']}")
else:
print(f"Errore {response.status_code}")
if __name__ == "__main__":
inspect_metadata()
+39
View File
@@ -0,0 +1,39 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def inspect_enums():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
query = """
query {
sched: __type(name: "SchedulingType") {
enumValues { name }
}
mode: __type(name: "ShareMode") {
enumValues { name }
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
print("\n--- VALORI SCHEDULING ---")
for v in data['data']['sched']['enumValues']:
print(v['name'])
print("\n--- VALORI SHARE MODE ---")
for v in data['data']['mode']['enumValues']:
print(v['name'])
else:
print(f"Errore {response.status_code}")
if __name__ == "__main__":
inspect_enums()
+52
View File
@@ -0,0 +1,52 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def inspect_enum():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Cerchiamo il tipo del campo 'type' in InstagramPostMetadataInput
query = """
query {
__type(name: "InstagramPostMetadataInput") {
inputFields {
name
type {
name
kind
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
data = response.json()
for f in data['data']['__type']['inputFields']:
if f['name'] == 'type':
print(f"Tipo del campo 'type': {f['type']['name']}")
# Ora ispezioniamo quell'ENUM
enum_query = f"""
query {{
__type(name: "{f['type']['name']}") {{
enumValues {{
name
}}
}}
}}
"""
res_enum = requests.post(url, headers=headers, json={'query': enum_query})
data_enum = res_enum.json()
print("Valori possibili:")
for val in data_enum['data']['__type']['enumValues']:
print(f"- {val['name']}")
if __name__ == "__main__":
inspect_enum()
+51
View File
@@ -0,0 +1,51 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def introspect():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Query per elencare tutte le mutation disponibili
query = """
query {
__schema {
mutationType {
fields {
name
args {
name
type {
name
kind
}
}
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
if 'errors' in data:
print(f"Errore: {data['errors']}")
else:
fields = data['data']['__schema']['mutationType']['fields']
print("\n--- MUTATION DISPONIBILI SU BUFFER ---")
for f in fields:
print(f"Nome: {f['name']}")
for arg in f['args']:
print(f" Arg: {arg['name']} | Type: {arg['type']['name']} ({arg['type']['kind']})")
else:
print(f"Errore {response.status_code}: {response.text}")
if __name__ == "__main__":
introspect()
+40
View File
@@ -0,0 +1,40 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def introspect_payload():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
query = """
query {
__type(name: "PostActionPayload") {
fields {
name
type {
name
kind
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
fields = data['data']['__type']['fields']
print("\n--- CAMPI DI POSTACTIONPAYLOAD ---")
for f in fields:
print(f"Campo: {f['name']} | Tipo: {f['type']['name']}")
else:
print(f"Errore {response.status_code}")
if __name__ == "__main__":
introspect_payload()
+56
View File
@@ -0,0 +1,56 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def list_profiles():
token = os.getenv("BUFFER_ACCESS_TOKEN")
if not token:
print("Errore: BUFFER_ACCESS_TOKEN non trovato nel file .env")
return
# Nuova URL per GraphQL API
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Query GraphQL per ottenere l'account e le organizzazioni
query = """
query {
account {
organizations {
id
name
channels {
id
service
name
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
if 'errors' in data:
print(f"Errore GraphQL: {data['errors']}")
return
organizations = data['data']['account']['organizations']
print("\n--- ORGANIZZAZIONI E CANALI BUFFER ---")
for org in organizations:
print(f"\nOrganizzazione: {org['name']} (ID: {org['id']})")
for channel in org['channels']:
print(f" - ID Canale: {channel['id']} | Social: {channel['service']} | Nome: {channel['name']}")
print("---------------------------------------\n")
else:
print(f"Errore API Buffer: {response.status_code} - {response.text}")
if __name__ == "__main__":
list_profiles()
+141
View File
@@ -0,0 +1,141 @@
import os
import sys
import json
import asyncio
from pathlib import Path
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
async def fetch_mcp_details(api_key, api_secret=None):
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
url = "https://mcp.higgsfield.ai/mcp"
# MCP server accepts Bearer tokens (unlike REST API which uses "Key" format)
# Note: MCP server is designed for OAuth browser auth, API keys provide limited access
headers = {
"Authorization": f"Bearer {api_key}"
}
print(f"🔌 Connecting to Higgsfield hosted MCP server ({url})...")
try:
async with streamablehttp_client(url, headers=headers) as (read, write, _):
async with ClientSession(read, write) as session:
print(" Initializing MCP session...")
await session.initialize()
print("✅ Session initialized successfully!")
# 1. Query available tools
print("\n🔍 Fetching available tools...")
tools_res = await session.list_tools()
tools = tools_res.tools
print(f" Found {len(tools)} tools:")
for t in tools:
desc = t.description[:80] if t.description else "No description"
print(f" 🛠️ Tool: {t.name} - {desc}...")
# 2. Query available resources
print("\n🔍 Fetching available resources...")
resources_res = await session.list_resources()
resources = resources_res.resources
print(f" Found {len(resources)} resources:")
for r in resources:
print(f" 🔗 Resource: {r.uri} - {r.name}")
# 3. Pre-flight check: verify API key works by calling balance
print("\n🔑 Verifying API key validity...")
try:
preflight = await session.call_tool("balance", {})
preflight_text = preflight.content[0].text if preflight.content else ""
if "something went wrong" in preflight_text.lower() or "error" in preflight_text.lower():
print(f" ⚠️ API key may be invalid or expired. Server response: {preflight_text}")
print(" 💡 Tip: Log in to https://higgsfield.ai and regenerate your API key.")
print(" The MCP connection works, but tool calls are failing server-side.")
else:
print(f" ✅ API key verified. Account info: {preflight_text}")
except Exception as e:
print(f" ⚠️ Pre-flight check failed: {e}")
# 4. Try to list characters via the show_characters tool
print("\n📖 Fetching Soul Characters via show_characters tool...")
char_tools = [t for t in tools if "character" in t.name.lower() or "soul" in t.name.lower()]
if char_tools:
for t in char_tools:
print(f" Calling tool: {t.name}...")
try:
result = await session.call_tool(t.name, {"action": "list", "status": "ready", "size": 50})
print(f"✅ Tool result:")
print("-" * 50)
if result.content:
text = result.content[0].text
try:
parsed = json.loads(text)
if isinstance(parsed, list):
for char in parsed:
char_id = char.get("id") or char.get("uuid") or char.get("soul_id")
name = char.get("name") or "Unnamed Character"
status = char.get("status", "unknown")
print(f"🆔 Soul ID: {char_id}")
print(f"👤 Name : {name}")
print(f"📊 Status : {status}")
print("-" * 50)
elif isinstance(parsed, dict):
items = parsed.get("characters") or parsed.get("items") or parsed.get("data") or [parsed]
if isinstance(items, list):
for char in items:
char_id = char.get("id") or char.get("uuid") or char.get("soul_id")
name = char.get("name") or "Unnamed Character"
status = char.get("status", "unknown")
print(f"🆔 Soul ID: {char_id}")
print(f"👤 Name : {name}")
print(f"📊 Status : {status}")
print("-" * 50)
else:
print(json.dumps(parsed, indent=2))
else:
print(text)
except json.JSONDecodeError:
print(text)
else:
print(" (No content returned)")
print("-" * 50)
except Exception as tool_err:
print(f" ⚠️ Could not invoke tool {t.name}: {tool_err}")
else:
print(" ⚠️ No character-related tools found on the server.")
# 4. Also check for any other listing tools
list_tools = [t for t in tools if "list" in t.name.lower() and t not in char_tools]
if list_tools:
print("\n💡 Other listing tools available:")
for t in list_tools:
desc = t.description[:100] if t.description else "No description"
print(f" 🛠️ {t.name}: {desc}")
except Exception as e:
print(f"❌ MCP Connection Error: {e}", file=sys.stderr)
import traceback
traceback.print_exc()
sys.exit(1)
def main():
api_key = os.getenv("HIGGSFIELD_API_KEY")
api_secret = os.getenv("HIGGSFIELD_SECRET")
if not api_key or api_key == "your_higgsfield_api_key_here":
print("❌ Error: HIGGSFIELD_API_KEY is not configured in .env.")
print(" Please add your real token to .env to authenticate.", file=sys.stderr)
sys.exit(1)
if not api_secret:
print("⚠️ Warning: HIGGSFIELD_SECRET is not set. Some API calls may fail.")
asyncio.run(fetch_mcp_details(api_key, api_secret))
if __name__ == "__main__":
main()
+11
View File
@@ -0,0 +1,11 @@
import os
import google.generativeai as genai
from dotenv import load_dotenv
load_dotenv()
genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
print("Modelli disponibili:")
for m in genai.list_models():
if 'generateContent' in m.supported_generation_methods:
print(m.name)
+49
View File
@@ -0,0 +1,49 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def list_profiles():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
query = """
query {
account {
id
email
organizations {
id
name
channels(input: {}) {
id
name
service
}
}
}
}
"""
response = requests.post(url, headers=headers, json={'query': query})
if response.status_code == 200:
data = response.json()
acc = data.get('data', {}).get('account')
if acc:
print(f"\nAccount: {acc['email']} ({acc['id']})")
for org in acc['organizations']:
print(f"\n--- ORGANIZZAZIONE: {org['name']} ({org['id']}) ---")
for c in org['channels']:
print(f"ID: {c['id']} | Servizio: {c['service']} | Nome: {c['name']}")
else:
print("Account non trovato:", data)
else:
print(f"Errore {response.status_code}: {response.text}")
if __name__ == "__main__":
list_profiles()
+99
View File
@@ -0,0 +1,99 @@
import os
import json
import time
import asyncio
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from telegram import Bot
from agents import SocialAgents
from database import Database
from dotenv import load_dotenv
load_dotenv()
class SchedulerService:
def __init__(self):
self.db = Database()
self.agents = SocialAgents()
self.bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
self.chat_id = os.getenv("TELEGRAM_CHAT_ID")
self.bot = Bot(token=self.bot_token) if self.bot_token else None
async def send_notification(self, message):
if self.bot and self.chat_id:
try:
await self.bot.send_message(chat_id=self.chat_id, text=message)
except Exception as e:
print(f"Errore invio Telegram: {e}")
def run_single_artist_job(self, artist):
print(f"[{datetime.now()}] Avvio generazione per {artist['name']}...")
try:
# Sincronizzazione automatica da Drive e locale
os.system("python src/cloud_sync.py")
os.system("python src/sync_assets.py")
# Validazione materiale prima della generazione
audio = self.db.get_audio_for_artist(artist['id'])
image = self.db.get_unused_image(artist['id'])
if not audio or not image:
msg = f"⚠️ Materiale mancante per {artist['name']}. Caricare MP3 e Foto!"
print(msg)
asyncio.run(self.send_notification(msg))
return
results = self.agents.run_for_artist(artist)
for result in results:
if 'error' not in result:
self.db.save_draft(
artist_id=artist['id'],
title=result['title'],
caption=result['caption'],
hashtags=result['hashtags'],
image_path=result['image_path'],
image_url=result.get('image_url'),
audio_analysis=result['audio_analysis'],
video_url=result.get('video_url')
)
asyncio.run(self.send_notification(f"✅ Post pronti per {artist['name']} (Foto + Video)!"))
except Exception as e:
err = f"❌ Errore per {artist['name']}: {str(e)}"
print(err)
asyncio.run(self.send_notification(err))
def start(self):
scheduler = BackgroundScheduler()
with open('config.json', 'r') as f:
config = json.load(f)
artists = config['artists']
for artist in artists:
s_time = artist.get('schedule_time', '09:00')
try:
hour, minute = map(int, s_time.split(':'))
scheduler.add_job(
self.run_single_artist_job,
'cron',
hour=hour,
minute=minute,
args=[artist],
id=f"job_{artist['id'].replace(' ', '_')}"
)
print(f"Pianificato {artist['name']} alle {s_time}")
except Exception as e:
print(f"Errore pianificazione per {artist['name']}: {e}")
scheduler.start()
print("Scheduler avviato.")
try:
while True:
time.sleep(60)
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
if __name__ == "__main__":
service = SchedulerService()
service.start()
+35
View File
@@ -0,0 +1,35 @@
import os
import json
from database import Database
def sync():
db = Database()
# Load config to get artists
with open('config.json', 'r') as f:
config = json.load(f)
artists = [a['id'] for a in config['artists']]
base_path = "data"
for artist_id in artists:
# Sync Images
img_dir = os.path.join(base_path, "images", artist_id)
if os.path.exists(img_dir):
for file in os.listdir(img_dir):
if file.lower().endswith(('.png', '.jpg', '.jpeg', '.webp')):
file_path = os.path.join(img_dir, file)
db.add_media(file_path, 'image', artist_id)
# Sync Audio
audio_dir = os.path.join(base_path, "audio", artist_id)
if os.path.exists(audio_dir):
for file in os.listdir(audio_dir):
if file.lower().endswith(('.mp3', '.wav', '.m4a')):
file_path = os.path.join(audio_dir, file)
db.add_media(file_path, 'audio', artist_id)
print("Sincronizzazione completata.")
if __name__ == "__main__":
sync()
+49
View File
@@ -0,0 +1,49 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def test_publish():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
mutation = """
mutation ($input: CreatePostInput!) {
createPost(input: $input) {
... on PostActionSuccess {
post {
id
}
}
... on MutationError {
message
}
}
}
"""
variables = {
"input": {
"channelId": "69f9ea855c4c051afa117baa",
"text": "Test di pubblicazione via API (In Coda) 🎼🚀",
"schedulingType": "automatic",
"mode": "addToQueue"
}
}
print("Inviando TEST REALE (In Coda) a Buffer...")
response = requests.post(url, headers=headers, json={
'query': mutation,
'variables': variables
})
print(f"Status Code: {response.status_code}")
print(f"Risposta: {response.text}")
if __name__ == "__main__":
test_publish()
+51
View File
@@ -0,0 +1,51 @@
import os
import requests
from dotenv import load_dotenv
load_dotenv()
def test_draft():
token = os.getenv("BUFFER_ACCESS_TOKEN")
url = "https://api.buffer.com/graphql"
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# Canale Instagram
channel_id = "69f9ea855c4c051afa117baa"
mutation = """
mutation ($input: CreatePostInput!) {
createPost(input: $input) {
... on PostActionSuccess {
post {
id
text
}
}
... on MutationError {
message
}
}
}
"""
variables = {
"input": {
"channelId": channel_id,
"text": "TEST BOZZA DAL BOT - " + os.popen("date").read().strip(),
"schedulingType": "automatic",
"mode": "addToQueue",
"saveToDraft": True
}
}
response = requests.post(url, headers=headers, json={'query': mutation, 'variables': variables})
if response.status_code == 200:
print("Risposta Buffer:", response.json())
else:
print(f"Errore {response.status_code}: {response.text}")
if __name__ == "__main__":
test_draft()
+28
View File
@@ -0,0 +1,28 @@
import json
from agents import SocialAgents
from database import Database
def test():
db = Database()
agents = SocialAgents()
with open('config.json', 'r') as f:
config = json.load(f)
# Cerchiamo l'artista 2
artist = next((a for a in config['artists'] if a['id'] == 'artista_2'), None)
if artist:
print(f"--- Inizio generazione per {artist['name']} ---")
result = agents.run_for_artist(artist)
print("\n--- RISULTATO GENERAZIONE ---")
print(f"IMMAGINE: {result['image_path']}")
print(f"\nANALISI AUDIO:\n{result['audio_analysis']}")
print(f"\nCAPTION:\n{result['caption']}")
print(f"\nHASHTAGS:\n{result['hashtags']}")
print("-----------------------------")
else:
print("Artista 2 non trovato in config.json")
if __name__ == "__main__":
test()
+184
View File
@@ -0,0 +1,184 @@
import os
import subprocess
import requests
import json
import time
class VideoGenerator:
def __init__(self, cache_dir="data/cache"):
self.cache_dir = cache_dir
os.makedirs(self.cache_dir, exist_ok=True)
def generate_video(self, media_paths, audio_path, title, start_time=0, output_path=None, bpm=120, focus_points=None):
if isinstance(media_paths, str):
media_paths = [media_paths]
if not output_path:
output_path = os.path.join(self.cache_dir, "final_video.mp4")
# Assicuriamoci che la cartella di destinazione esista
os.makedirs(os.path.dirname(output_path), exist_ok=True)
if os.path.exists(output_path):
os.remove(output_path)
# Assicuriamoci che focus_points sia una lista della stessa lunghezza di media_paths
if not focus_points:
focus_points = [None] * len(media_paths)
safe_title = title.replace(":", "\\:").replace("'", "").replace("%", "")
# Calcolo tempi basato su BPM (Sincronizzazione)
beat_duration = 60.0 / bpm
num_assets = len(media_paths)
# Costruiamo il comando FFmpeg dinamico
inputs = []
for asset in media_paths:
inputs.extend(['-i', asset])
filter_complex = ""
import random
for i, asset in enumerate(media_paths):
is_video = asset.lower().endswith(('.mp4', '.mov', '.avi', '.mkv'))
focus = focus_points[i] if i < len(focus_points) else None
# Ritmo basato su multipli del beat
beats_per_clip = random.choice([2, 4])
if random.random() > 0.8: beats_per_clip = 8
d_img_curr = beat_duration * beats_per_clip
d_frames = int(d_img_curr * 25)
# Effetti comuni
clean_effects = [
"eq=contrast=1.2:saturation=1.5",
"colorbalance=rs=0.2:gs=0.1:bs=0.2,eq=contrast=1.3:saturation=1.4",
"eq=brightness=0.05:contrast=1.4:saturation=1.3",
"unsharp=5:5:1.0:5:5:0.0"
]
effect = random.choice(clean_effects)
if random.random() > 0.7:
effect += ",rgbashift=rh=2:rv=2:gh=-2:gv=-2"
flash = f"fade=t=in:st=0:d=0.2:color=white,"
if not is_video:
# LOGICA IMMAGINE: Zoompan
zoom_speed = 0.005 if beats_per_clip < 4 else 0.002
if focus:
# Se abbiamo il focus dall'AI, puntiamo lì (coordinate 0-100)
fx = focus.get('x', 50) / 100.0
fy = focus.get('y', 40) / 100.0 # Leggermente più su del centro se non specificato
# Centriamo il pan sul punto di interesse
target_x = f"iw*{fx}-(iw/zoom/2)"
target_y = f"ih*{fy}-(ih/zoom/2)"
else:
# Altrimenti movimento casuale classico
dir_x = random.uniform(-0.5, 0.5)
dir_y = random.uniform(-0.3, 0.3)
target_x = f"iw/2-(iw/zoom/2)+({dir_x}*on)"
target_y = f"ih/2-(ih/zoom/2)+({dir_y}*on)"
filter_complex += f"[{i}:v]scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920," \
f"zoompan=z='min(max(zoom,1.0)+{zoom_speed},1.5)':" \
f"x='{target_x}':" \
f"y='{target_y}':" \
f"d={d_frames}:s=1080x1920," \
f"{flash}{effect},setsar=1[v{i}];"
else:
# LOGICA VIDEO: Trim e Scale
# Prendiamo uno spezzone casuale (assumiamo video > 5s per sicurezza)
start_trim = random.uniform(0, 2.0)
filter_complex += f"[{i}:v]scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920," \
f"trim=start={start_trim}:duration={d_img_curr},setpts=PTS-STARTPTS," \
f"fps=25,{flash}{effect},setsar=1[v{i}];"
concat_inputs = ""
for i in range(num_assets):
concat_inputs += f"[v{i}]"
filter_complex += f"{concat_inputs}concat=n={num_assets}:v=1:a=0[outv]"
temp_block_path = os.path.join(self.cache_dir, "temp_block.mp4")
cmd_block = ['ffmpeg', '-y'] + inputs + [
'-filter_complex', filter_complex,
'-map', '[outv]',
'-c:v', 'libx264', '-crf', '23', '-preset', 'fast', '-pix_fmt', 'yuv420p',
temp_block_path
]
print(f"🛠️ FFmpeg: Generazione blocco base (Filtro complesso)...")
res1 = subprocess.run(cmd_block, capture_output=True, text=True)
if res1.returncode != 0:
print(f"❌ ERRORE FFMPEG BLOCK: {res1.stderr}")
raise Exception("FFmpeg failed to create base block")
print(f"🛠️ FFmpeg: Assemblaggio finale (Loop + Audio)...")
cmd_final = [
'ffmpeg', '-y',
'-stream_loop', '-1', '-i', temp_block_path,
'-ss', str(start_time), '-i', audio_path,
'-c:v', 'copy',
'-c:a', 'aac', '-b:a', '192k',
'-t', '30', '-shortest',
output_path
]
res2 = subprocess.run(cmd_final, capture_output=True, text=True)
if os.path.exists(temp_block_path):
os.remove(temp_block_path)
if res2.returncode != 0:
print(f"ERRORE FFMPEG FINAL: {res2.stderr}")
raise Exception("FFmpeg failed to assemble final video")
return output_path
def generate_thumbnail(self, video_path, output_path=None):
"""Genera una miniatura locale dal video generato"""
if not output_path:
output_path = video_path.replace(".mp4", ".jpg")
if os.path.exists(output_path): return output_path
print(f"🖼️ Generazione miniatura locale: {output_path}...")
cmd = [
'ffmpeg', '-y', '-ss', '00:00:01',
'-i', video_path,
'-vframes', '1', '-q:v', '2',
output_path
]
res = subprocess.run(cmd, capture_output=True)
if res.returncode == 0:
return output_path
return None
def upload_to_ephemeral(self, file_path):
"""Carica su host temporanei per ottenere un URL pubblico per Buffer"""
# Tentativo 1: BashUpload (Molto semplice e diretto)
print(f"Tentativo upload su BashUpload...")
try:
filename = os.path.basename(file_path)
with open(file_path, 'rb') as f:
response = requests.put(f'https://bashupload.com/{filename}', data=f, timeout=30)
if response.status_code == 200:
# BashUpload restituisce l'URL nel corpo della risposta
for line in response.text.split('\n'):
if 'https://bashupload.com/' in line:
return line.strip().split(' ')[-1]
except Exception as e: print(f"BashUpload errore: {e}")
# Tentativo 2: Uguu.se
print(f"Tentativo upload su Uguu.se...")
try:
with open(file_path, 'rb') as f:
response = requests.post('https://uguu.se/upload.php', files={'files[]': f}, timeout=30)
if response.status_code == 200:
return response.json()['files'][0]['url']
except Exception as e: print(f"Uguu.se errore: {e}")
return None