513 lines
16 KiB
Python
Executable File
513 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.request
|
|
import urllib.error
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import psycopg
|
|
|
|
|
|
def env(name: str, default: str = "") -> str:
|
|
return os.getenv(name, default).strip()
|
|
|
|
|
|
PSQL_DATABASE_URL = env("PSQL_DATABASE_URL")
|
|
OPENROUTER_API_KEY = env("OPENROUTER_API_KEY")
|
|
OPENROUTER_URL = env("OPENROUTER_URL", "https://openrouter.ai/api/v1/chat/completions")
|
|
OPENROUTER_MODEL = env("EXTRACTION_MODEL", env("OPENROUTER_MODEL", "qwen/qwen3-30b-a3b"))
|
|
|
|
|
|
def extract_first_json_object(raw: str) -> str:
|
|
s = str(raw or "").strip()
|
|
|
|
if s.startswith("```"):
|
|
lines = s.splitlines()
|
|
if lines and lines[0].strip().startswith("```"):
|
|
lines = lines[1:]
|
|
if lines and lines[-1].strip().startswith("```"):
|
|
lines = lines[:-1]
|
|
s = "\n".join(lines).strip()
|
|
|
|
start = s.find("{")
|
|
if start == -1:
|
|
raise ValueError(f"no JSON object found: {s[:300]}")
|
|
|
|
in_string = False
|
|
escaped = False
|
|
depth = 0
|
|
|
|
for i in range(start, len(s)):
|
|
ch = s[i]
|
|
|
|
if escaped:
|
|
escaped = False
|
|
continue
|
|
|
|
if ch == "\\":
|
|
escaped = True
|
|
continue
|
|
|
|
if ch == '"':
|
|
in_string = not in_string
|
|
continue
|
|
|
|
if in_string:
|
|
continue
|
|
|
|
if ch == "{":
|
|
depth += 1
|
|
elif ch == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return s[start:i + 1]
|
|
|
|
raise ValueError(f"incomplete JSON object: {s[:500]}")
|
|
|
|
|
|
def parse_json(raw: str) -> Dict[str, Any]:
|
|
try:
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return json.loads(extract_first_json_object(raw))
|
|
|
|
|
|
def strip_html(text: str) -> str:
|
|
text = re.sub(r"<br\s*/?>", "\n", text or "", flags=re.I)
|
|
text = re.sub(r"</p\s*>", "\n", text, flags=re.I)
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
text = text.replace(" ", " ")
|
|
text = text.replace("&", "&")
|
|
text = text.replace(""", '"')
|
|
text = text.replace("'", "'")
|
|
return re.sub(r"[ \t]+", " ", text).strip()
|
|
|
|
|
|
def remove_quoted_text(text: str) -> str:
|
|
if not text:
|
|
return ""
|
|
|
|
markers = [
|
|
"\nÀs ",
|
|
"\nEm ",
|
|
"\nOn ",
|
|
"\n-----Original Message-----",
|
|
"\nDe:",
|
|
"\nFrom:",
|
|
]
|
|
|
|
cut = len(text)
|
|
for marker in markers:
|
|
idx = text.find(marker)
|
|
if idx != -1:
|
|
cut = min(cut, idx)
|
|
|
|
lines = []
|
|
for line in text[:cut].splitlines():
|
|
if line.strip().startswith(">"):
|
|
continue
|
|
lines.append(line)
|
|
|
|
return "\n".join(lines).strip()
|
|
|
|
|
|
def extract_message_content(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
msg = payload.get("message") or payload.get("messages") or {}
|
|
|
|
if isinstance(msg, list):
|
|
msg = msg[0] if msg else {}
|
|
|
|
if not isinstance(msg, dict):
|
|
return None
|
|
|
|
content = msg.get("content") or payload.get("content") or ""
|
|
message_type = str(msg.get("message_type") or payload.get("message_type") or "").lower()
|
|
private = bool(msg.get("private") or payload.get("private"))
|
|
|
|
if not content:
|
|
return None
|
|
|
|
return {
|
|
"content": remove_quoted_text(strip_html(str(content))),
|
|
"message_type": message_type,
|
|
"private": private,
|
|
"id": str(msg.get("id") or payload.get("id") or ""),
|
|
}
|
|
|
|
|
|
def fetch_task(conn, task_id: Optional[str], conversation_id: Optional[str]) -> Dict[str, Any]:
|
|
if task_id:
|
|
sql = """
|
|
select id::text, conversation_id, contact_id, action_code, route, status, action, note
|
|
from tasks
|
|
where id = %s
|
|
limit 1
|
|
"""
|
|
params = (task_id,)
|
|
else:
|
|
sql = """
|
|
select id::text, conversation_id, contact_id, action_code, route, status, action, note
|
|
from tasks
|
|
where conversation_id = %s
|
|
order by created_at desc
|
|
limit 1
|
|
"""
|
|
params = (conversation_id,)
|
|
|
|
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
|
|
cur.execute(sql, params)
|
|
row = cur.fetchone()
|
|
|
|
if not row:
|
|
raise SystemExit("ERRO: task não encontrada.")
|
|
|
|
return dict(row)
|
|
|
|
|
|
def fetch_conversation_messages(conn, conversation_id: str, limit: int = 20) -> List[Dict[str, Any]]:
|
|
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
select created_at, payload
|
|
from raw_events
|
|
where source_system = 'chatwoot'
|
|
and conversation_id = %s
|
|
order by created_at asc
|
|
limit %s
|
|
""",
|
|
(conversation_id, limit),
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
messages = []
|
|
for row in rows:
|
|
payload = row.get("payload") or {}
|
|
if isinstance(payload, str):
|
|
try:
|
|
payload = json.loads(payload)
|
|
except Exception:
|
|
payload = {}
|
|
|
|
msg = extract_message_content(payload)
|
|
if not msg:
|
|
continue
|
|
|
|
# Para extração, manter incoming e outgoing públicos, ignorar notas privadas.
|
|
if msg["private"]:
|
|
continue
|
|
|
|
messages.append({
|
|
"created_at": str(row["created_at"]),
|
|
**msg,
|
|
})
|
|
|
|
return messages
|
|
|
|
|
|
def build_prompt(task: Dict[str, Any], messages: List[Dict[str, Any]], prep_type: str) -> List[Dict[str, str]]:
|
|
conversation_text = "\n\n".join(
|
|
f"[{m['created_at']}] {m.get('message_type') or 'message'}:\n{m['content']}"
|
|
for m in messages
|
|
if m.get("content")
|
|
)
|
|
|
|
if prep_type == "proforma":
|
|
objective = """
|
|
Objetivo: preparar dados para emitir fatura pró-forma.
|
|
Extrai apenas dados relevantes para faturação/proforma:
|
|
cliente, empresa, email, telefone, NIF, morada fiscal, produto, quantidade, preço, condições comerciais e dados em falta para emitir a pró-forma.
|
|
|
|
Para prep_type=proforma:
|
|
- NÃO peças morada de entrega, destinatário ou telefone para transportadora, exceto se a conversa indicar que são necessários para a pró-forma.
|
|
- Morada de entrega/recolha pertence à fase de envio/recolha, não à fase de pró-forma.
|
|
- Se já houver NIF, morada fiscal, nome/empresa de faturação, email, produto e preço, missing_fields deve ser [].
|
|
"""
|
|
elif prep_type == "shipment":
|
|
objective = """
|
|
Objetivo: preparar envio.
|
|
Extrai apenas dados relevantes para logística de entrega:
|
|
morada de entrega, contacto no local, telefone, produto/equipamento, quantidade, instruções, estado do pagamento e dados em falta.
|
|
|
|
Para prep_type=shipment:
|
|
- Usa shipment.delivery_address para a morada de entrega.
|
|
- Usa shipment.recipient_name e shipment.recipient_phone para contacto da entrega.
|
|
- Os campos obrigatórios são morada de entrega, contacto, telefone, produto/equipamento, quantidade e estado do pagamento.
|
|
- NÃO coloques customer.tax_id, billing.tax_id, billing_address ou customer.email em missing_fields, exceto se forem explicitamente necessários para a transportadora.
|
|
- Se já existir morada de entrega, contacto e telefone, não peças esses dados novamente.
|
|
- Se faltar produto, quantidade ou comprovativo/estado de pagamento, pede apenas esses dados.
|
|
- A suggested_reply deve falar em envio/entrega, nunca em recolha.
|
|
"""
|
|
elif prep_type == "pickup":
|
|
objective = """
|
|
Objetivo: preparar recolha.
|
|
Extrai apenas dados relevantes para recolha ou assistência logística:
|
|
morada de recolha, contacto no local, telefone, produto/equipamento a recolher, motivo/instruções, data preferida e dados em falta.
|
|
|
|
Para prep_type=pickup:
|
|
- Usa shipment.pickup_address para a morada de recolha.
|
|
- Usa shipment.recipient_name e shipment.recipient_phone para a pessoa de contacto da recolha.
|
|
- Se a conversa só tiver uma morada e o objetivo é recolha, coloca essa morada em shipment.pickup_address, não em shipment.delivery_address.
|
|
- NÃO coloques sale.total_estimate, customer.tax_id, billing.tax_id, billing_address ou customer.email em missing_fields.
|
|
- Os campos importantes são pickup_address, recipient_name, recipient_phone, produto/equipamento, motivo/instruções da recolha.
|
|
- payment.status só é obrigatório se a tarefa for claramente sobre pagamento ou envio após pagamento.
|
|
- A suggested_reply deve falar em recolha, nunca em envio.
|
|
"""
|
|
else:
|
|
objective = """
|
|
Objetivo: preparar execução operacional da tarefa.
|
|
Extrai dados úteis para a ação, dados em falta e resposta sugerida.
|
|
"""
|
|
|
|
system = f"""
|
|
És o ClientFlow Sales Assistant.
|
|
A tua função é extrair dados operacionais de conversas B2B para ajudar a preparar pró-forma, fatura, envio ou recolha.
|
|
|
|
Regras:
|
|
- Devolve apenas JSON puro.
|
|
- Não inventes dados.
|
|
- Se um dado não existir, usa null.
|
|
- Ignora texto citado antigo, dados da BLIF, IBANs e assinaturas da BLIF.
|
|
- Não associes IBAN ao cliente.
|
|
- Distingue morada fiscal de morada de entrega/recolha.
|
|
- Usa evidências curtas.
|
|
- Se faltar dado necessário, coloca em missing_fields.
|
|
- suggested_reply deve ser uma resposta curta em português para pedir dados em falta ou indicar o próximo passo.
|
|
- Não digas que a pró-forma/fatura/envio já foi emitida, enviada, agendada ou concluída.
|
|
- O assistente apenas prepara dados para revisão; usa linguagem como "vamos preparar", "podemos avançar", "dados suficientes para preparar".
|
|
"""
|
|
|
|
user = f"""
|
|
{objective}
|
|
|
|
Tarefa:
|
|
action_code: {task.get('action_code')}
|
|
route: {task.get('route')}
|
|
action: {task.get('action')}
|
|
note: {task.get('note')}
|
|
|
|
Conversa:
|
|
{conversation_text}
|
|
|
|
Schema obrigatório:
|
|
{{
|
|
"customer": {{
|
|
"name": null,
|
|
"company": null,
|
|
"email": null,
|
|
"phone": null,
|
|
"tax_id": null
|
|
}},
|
|
"billing": {{
|
|
"billing_name": null,
|
|
"tax_id": null,
|
|
"billing_address": null,
|
|
"billing_email": null
|
|
}},
|
|
"sale": {{
|
|
"products": [
|
|
{{
|
|
"name": null,
|
|
"description": null,
|
|
"quantity": null,
|
|
"unit_price": null,
|
|
"currency": "EUR"
|
|
}}
|
|
],
|
|
"total_estimate": null,
|
|
"commercial_terms": null
|
|
}},
|
|
"payment": {{
|
|
"status": null,
|
|
"proof_mentioned": false
|
|
}},
|
|
"shipment": {{
|
|
"delivery_address": null,
|
|
"pickup_address": null,
|
|
"recipient_name": null,
|
|
"recipient_phone": null,
|
|
"instructions": null
|
|
}},
|
|
"missing_fields": [],
|
|
"suggested_reply": "",
|
|
"confidence": 0.0,
|
|
"evidence": []
|
|
}}
|
|
"""
|
|
|
|
return [
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": user},
|
|
]
|
|
|
|
|
|
def call_openrouter(messages: List[Dict[str, str]]) -> Tuple[Dict[str, Any], Dict[str, Any], str]:
|
|
body = {
|
|
"model": OPENROUTER_MODEL,
|
|
"messages": messages,
|
|
"temperature": 0,
|
|
}
|
|
|
|
req = urllib.request.Request(
|
|
OPENROUTER_URL,
|
|
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
|
|
headers={
|
|
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
"HTTP-Referer": "https://clientflow.blif.pt",
|
|
"X-Title": "ClientFlow",
|
|
},
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|
parsed = json.loads(raw)
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"OpenRouter HTTP {e.code}: {raw[:1000]}")
|
|
|
|
content = parsed["choices"][0]["message"].get("content") or "{}"
|
|
extracted = parse_json(content)
|
|
return extracted, parsed, content
|
|
|
|
|
|
def save_preparation(conn, task: Dict[str, Any], prep_type: str, extracted: Dict[str, Any], raw_response: Dict[str, Any]) -> str:
|
|
usage = raw_response.get("usage") or {}
|
|
model = raw_response.get("model") or OPENROUTER_MODEL
|
|
provider = raw_response.get("provider") or raw_response.get("provider_name")
|
|
|
|
missing_fields = extracted.get("missing_fields") or []
|
|
suggested_reply = extracted.get("suggested_reply") or ""
|
|
confidence = extracted.get("confidence")
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
insert into task_preparations (
|
|
task_id,
|
|
conversation_id,
|
|
contact_id,
|
|
prep_type,
|
|
status,
|
|
extracted_data,
|
|
missing_fields,
|
|
suggested_reply,
|
|
confidence,
|
|
model,
|
|
provider,
|
|
total_tokens,
|
|
cost,
|
|
raw_response
|
|
)
|
|
values (
|
|
%s, %s, %s, %s, 'draft',
|
|
%s::jsonb,
|
|
%s::jsonb,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s::jsonb
|
|
)
|
|
returning id::text
|
|
""",
|
|
(
|
|
task["id"],
|
|
task["conversation_id"],
|
|
task.get("contact_id"),
|
|
prep_type,
|
|
json.dumps(extracted, ensure_ascii=False),
|
|
json.dumps(missing_fields, ensure_ascii=False),
|
|
suggested_reply,
|
|
confidence,
|
|
model,
|
|
provider,
|
|
int(usage.get("total_tokens") or 0),
|
|
usage.get("cost") or 0,
|
|
json.dumps(raw_response, ensure_ascii=False),
|
|
),
|
|
)
|
|
prep_id = cur.fetchone()[0]
|
|
|
|
conn.commit()
|
|
return prep_id
|
|
|
|
|
|
def run_preparation(
|
|
*,
|
|
task_id: Optional[str] = None,
|
|
conversation_id: Optional[str] = None,
|
|
prep_type: str,
|
|
database_url: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
db_url = database_url or PSQL_DATABASE_URL
|
|
if not task_id and not conversation_id:
|
|
raise ValueError("Usa task_id ou conversation_id.")
|
|
if not db_url:
|
|
raise RuntimeError("PSQL_DATABASE_URL/DATABASE_URL não definida.")
|
|
if not OPENROUTER_API_KEY:
|
|
raise RuntimeError("OPENROUTER_API_KEY não definida.")
|
|
|
|
with psycopg.connect(db_url) as conn:
|
|
task = fetch_task(conn, task_id, conversation_id)
|
|
messages = fetch_conversation_messages(conn, task["conversation_id"])
|
|
if not messages:
|
|
raise RuntimeError("Não encontrei mensagens públicas da conversa.")
|
|
prompt = build_prompt(task, messages, prep_type)
|
|
extracted, raw_response, raw_content = call_openrouter(prompt)
|
|
prep_id = save_preparation(conn, task, prep_type, extracted, raw_response)
|
|
|
|
return {
|
|
"preparation_id": prep_id,
|
|
"task_id": task["id"],
|
|
"conversation_id": task["conversation_id"],
|
|
"prep_type": prep_type,
|
|
"extracted": extracted,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--task-id")
|
|
parser.add_argument("--conversation-id")
|
|
parser.add_argument("--type", choices=["proforma", "shipment", "pickup", "generic"], required=True)
|
|
args = parser.parse_args()
|
|
|
|
if not args.task_id and not args.conversation_id:
|
|
raise SystemExit("Usa --task-id ou --conversation-id.")
|
|
|
|
if not PSQL_DATABASE_URL:
|
|
raise SystemExit("PSQL_DATABASE_URL não definida.")
|
|
|
|
if not OPENROUTER_API_KEY:
|
|
raise SystemExit("OPENROUTER_API_KEY não definida.")
|
|
|
|
with psycopg.connect(PSQL_DATABASE_URL) as conn:
|
|
task = fetch_task(conn, args.task_id, args.conversation_id)
|
|
messages = fetch_conversation_messages(conn, task["conversation_id"])
|
|
|
|
if not messages:
|
|
raise SystemExit("ERRO: não encontrei mensagens públicas da conversa.")
|
|
|
|
prompt = build_prompt(task, messages, args.type)
|
|
extracted, raw_response, raw_content = call_openrouter(prompt)
|
|
prep_id = save_preparation(conn, task, args.type, extracted, raw_response)
|
|
|
|
print(f"OK: preparation_id={prep_id}")
|
|
print(f"task_id={task['id']}")
|
|
print(f"conversation_id={task['conversation_id']}")
|
|
print(f"prep_type={args.type}")
|
|
print("--- extracted")
|
|
print(json.dumps(extracted, ensure_ascii=False, indent=2))
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|