| Server IP : 54.37.205.81 / Your IP : 216.73.216.76 Web Server : nginx/1.22.1 System : Linux vps-249481fa 6.1.0-50-cloud-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.176-1 (2026-07-02) x86_64 User : debian ( 1000) PHP Version : 8.2.32 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : OFF Directory : /opt/ccp-ai-gateway/ |
Upload File : |
import os
import json
import time
import secrets
import subprocess
import re
from datetime import datetime, timezone
import hashlib, secrets
from flask import Flask, request, jsonify, make_response, send_file
from flask_cors import CORS
import requests
import stripe
# PDF deps
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import A4
from reportlab.lib.units import cm
# QR (optional but supported)
try:
import qrcode
HAS_QRCODE = True
except Exception:
qrcode = None
HAS_QRCODE = False
APP = Flask(__name__)
CORS(
APP,
resources={r"/*": {"origins": [
"https://admin.copyrightchain.it",
"https://app.copyrightchain.it",
"https://copyrightchain.it",
]}},
supports_credentials=True,
allow_headers=["Content-Type", "Authorization", "X-CCP-Token"],
methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
)
# ================== ENV ==================
ADMIN_TOKEN = os.environ.get("CCP_AI_ADMIN_TOKEN", "")
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
TOWER_PASSWORD = os.environ.get("CCP_TOWER_PASSWORD", "")
SESSION_SECRET = os.environ.get("CCP_SESSION_SECRET", "")
DATA_DIR = os.environ.get("CCP_AI_DATA_DIR", "/opt/ccp-ai-gateway/data")
USERS_FILE = os.path.join(DATA_DIR, "users.json")
USAGE_LOG = os.path.join(DATA_DIR, "ai_usage.jsonl")
CERT_USAGE_LOG = os.path.join(DATA_DIR, "cert_usage.jsonl")
BALANCES_FILE = os.path.join(DATA_DIR, "balances.json")
AGENCY_PLANS_FILE = os.path.join(DATA_DIR, "agency_plans.json")
AGENCY_USAGE_LOG = os.path.join(DATA_DIR, "agency_usage.jsonl")
PRICING_FILE = os.path.join(DATA_DIR, "pricing.json")
# Public base URL (optional): e.g. https://copyrightchain.it/api/ai
PUBLIC_BASE_URL = os.environ.get("CCP_PUBLIC_BASE_URL", "").rstrip("/")
VERIFY_BASE_URL = os.environ.get("CCP_VERIFY_BASE_URL", "").rstrip("/")
STRIPE_SECRET_KEY = os.environ.get("CCP_STRIPE_SECRET_KEY", "").strip()
STRIPE_WEBHOOK_SECRET = os.environ.get("CCP_STRIPE_WEBHOOK_SECRET", "").strip()
STRIPE_SUCCESS_URL = os.environ.get("CCP_STRIPE_SUCCESS_URL", "").strip()
STRIPE_CANCEL_URL = os.environ.get("CCP_STRIPE_CANCEL_URL", "").strip()
WELCOME_PROMO_CODE = "WELCOM1"
WELCOME_PROMO_BONUS_ABO = 1
PROMO_USAGE_FILE = os.path.join(DATA_DIR, "promo_usage.json")
if STRIPE_SECRET_KEY:
stripe.api_key = STRIPE_SECRET_KEY
# Certificates folders
CERT_BASE_DIR = os.environ.get("CCP_CERT_BASE_DIR", os.path.join(DATA_DIR, "certificates", "base"))
CERT_BC_DIR = os.environ.get("CCP_CERT_BLOCKCHAIN_DIR", os.path.join(DATA_DIR, "certificates", "blockchain"))
CERT_TS_DIR = os.environ.get("CCP_CERT_TIMESTAMP_DIR", os.path.join(DATA_DIR, "certificates", "timestamp"))
OPENAPI_TIMESTAMP_ENABLED = os.environ.get("OPENAPI_TIMESTAMP_ENABLED", "0").strip().lower() in ("1", "true", "yes", "on")
OPENAPI_TIMESTAMP_ENV = os.environ.get("OPENAPI_TIMESTAMP_ENV", "sandbox").strip().lower()
OPENAPI_TIMESTAMP_TOKEN = os.environ.get("OPENAPI_TIMESTAMP_TOKEN", "").strip()
OPENAPI_TIMESTAMP_LOTTO_USERNAME = os.environ.get("OPENAPI_TIMESTAMP_LOTTO_USERNAME", "").strip()
OPENAPI_TIMESTAMP_LOTTO_PASSWORD = os.environ.get("OPENAPI_TIMESTAMP_LOTTO_PASSWORD", "").strip()
OPENAPI_TIMESTAMP_PROVIDER = os.environ.get("OPENAPI_TIMESTAMP_PROVIDER", "Openapi / InfoCert").strip()
OPENAPI_TIMESTAMP_TYPE = os.environ.get("OPENAPI_TIMESTAMP_TYPE", "infocert").strip().lower()
OPENAPI_TIMESTAMP_TIMEOUT = float(os.environ.get("OPENAPI_TIMESTAMP_TIMEOUT", "30"))
OTS_BIN = os.environ.get("CCP_OTS_BIN", "/home/debian/.local/bin/ots").strip() or "/home/debian/.local/bin/ots"
OTS_WORK_DIR = os.environ.get("CCP_OTS_WORK_DIR", os.path.join(DATA_DIR, "ots"))
ABO_CERT_FOOTER = 'ABO BANK Legal Office, Department of Intellectual Property and Copyright - Bank Association Offshore 1001 Third Avenue W, Suite 190, Bradenton, FL 34205 21. ABO BANK Harley Street, London, England, United Kingdom, W1G 9QD . Associazione Banche Offshore Corso Venezia 1 Milano Incorporata Codice Fiscale 92034960457'
# Stima costo EUR (solo reporting, non billing)
COST_OPENAI_EUR = float(os.environ.get("CCP_COST_OPENAI_EUR_PER_REQ", "0.01"))
COST_GEMINI_EUR = float(os.environ.get("CCP_COST_GEMINI_EUR_PER_REQ", "0.04"))
os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(CERT_BASE_DIR, exist_ok=True)
os.makedirs(CERT_BC_DIR, exist_ok=True)
os.makedirs(CERT_TS_DIR, exist_ok=True)
os.makedirs(OTS_WORK_DIR, exist_ok=True)
# ================== UTILS ==================
def _now_iso():
return datetime.now(timezone.utc).isoformat()
def _json_load(path, default):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return default
def _json_save(path, obj):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
os.replace(tmp, path)
def _append_jsonl(path, obj):
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(obj, ensure_ascii=False) + "\n")
def _draw_pdf_footer(c, generated_at_text=""):
import textwrap
text = c.beginText(2*cm, 2.2*cm)
text.setFont("Helvetica", 7)
text.setLeading(8)
for line in textwrap.wrap(ABO_CERT_FOOTER, width=115):
text.textLine(line)
c.drawText(text)
c.setFont("Helvetica-Oblique", 9)
c.drawString(2*cm, 1.2*cm, generated_at_text or f"Generated at (UTC): {_now_iso()}")
# ================== AUTH (admin header token) ==================
def require_admin_token(req):
def _env_key(name: str) -> str:
val = os.environ.get(name, "") or ""
if val:
return val.strip()
try:
with open("/opt/ccp-ai-gateway/.env", "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line.startswith(name + "="):
return line.split("=", 1)[1].strip().strip("\\\"'")
except Exception:
pass
return ""
allowed = [x for x in [ADMIN_TOKEN, _env_key("CCP_POSTAL_ADMIN_KEY")] if x]
if not allowed:
return True
candidates = [
req.headers.get("X-CCP-Token", ""),
req.headers.get("X-CCP-Admin-Key", ""),
str(req.headers.get("Authorization", "")).replace("Bearer ", "").strip(),
]
return any(candidate and candidate == token for candidate in candidates for token in allowed)
# ================== TOWER COOKIE SIGN ==================
def _sign(payload: dict) -> str:
raw = json.dumps(payload, separators=(",", ":"), sort_keys=True)
salt = secrets.token_urlsafe(12)
import hashlib
h = hashlib.sha256()
h.update(SESSION_SECRET.encode("utf-8"))
h.update(salt.encode("utf-8"))
h.update(raw.encode("utf-8"))
return f"{salt}.{h.hexdigest()}.{raw}"
def _verify(token: str):
try:
salt, digest, raw = token.split(".", 2)
import hashlib
h = hashlib.sha256()
h.update(SESSION_SECRET.encode("utf-8"))
h.update(salt.encode("utf-8"))
h.update(raw.encode("utf-8"))
if h.hexdigest() != digest:
return None
payload = json.loads(raw)
if payload.get("exp") and time.time() > float(payload["exp"]):
return None
return payload
except Exception:
return None
def tower_session():
tok = request.cookies.get("ccp_tower", "")
# cookie jar può salvarlo tra doppi apici e con virgole escaped (\054)
tok = (tok or "").strip('"').replace("\\054", ",")
if not tok or not SESSION_SECRET:
return None
payload = _verify(tok)
if not payload or payload.get("role") != "tower_admin":
return None
return payload
def _tower_guard():
if not (TOWER_PASSWORD and SESSION_SECRET and tower_session()):
return jsonify({"error": "unauthorized"}), 401
return None
# ================== USERS ==================
def ensure_users_file():
if not os.path.exists(USERS_FILE):
_json_save(USERS_FILE, {"users": []})
def find_user_by_token(tok: str):
tok = (tok or "").strip()
if not tok:
return None
ensure_users_file()
data = _json_load(USERS_FILE, {"users": []})
for u in data.get("users", []):
if (u.get("token") or "").strip() == tok:
return u
return None
def upsert_user(user: dict):
ensure_users_file()
data = _json_load(USERS_FILE, {"users": []})
users = data.get("users", [])
uid = str(user.get("id"))
found = False
for i, u in enumerate(users):
if str(u.get("id")) == uid:
users[i] = {**u, **user}
found = True
break
if not found:
users.append(user)
data["users"] = users
_json_save(USERS_FILE, data)
# ================== BALANCES (ABO) ==================
def ensure_balances_file():
if not os.path.exists(BALANCES_FILE):
_json_save(BALANCES_FILE, {"balances": {}})
def get_balance(uid: str) -> float:
ensure_balances_file()
data = _json_load(BALANCES_FILE, {"balances": {}})
b = data.get("balances", {}).get(str(uid), 0.0)
try:
return float(b)
except Exception:
return 0.0
def set_balance(uid: str, amount: float):
ensure_balances_file()
data = _json_load(BALANCES_FILE, {"balances": {}})
data.setdefault("balances", {})[str(uid)] = float(amount)
_json_save(BALANCES_FILE, data)
def add_balance(uid: str, delta: float) -> float:
cur = get_balance(uid)
newv = float(cur) + float(delta)
set_balance(uid, newv)
return newv
# ================== AGENCY CERTIFICATION CREDITS ==================
def ensure_agency_plans_file():
if not os.path.exists(AGENCY_PLANS_FILE):
_json_save(AGENCY_PLANS_FILE, {"plans": {}})
def _load_agency_plans():
ensure_agency_plans_file()
data = _json_load(AGENCY_PLANS_FILE, {"plans": {}})
if not isinstance(data, dict):
data = {"plans": {}}
if not isinstance(data.get("plans"), dict):
data["plans"] = {}
return data
def _save_agency_plans(data):
if not isinstance(data, dict):
data = {"plans": {}}
if not isinstance(data.get("plans"), dict):
data["plans"] = {}
_json_save(AGENCY_PLANS_FILE, data)
def _agency_int(value, default=0):
try:
return int(float(value))
except Exception:
return int(default)
def agency_cert_cost(op: str) -> int:
op = (op or "").strip().lower()
return {
"cert_base": 1,
"cert_blockchain": 2,
"cert_timestamp": 3,
}.get(op, 0)
def get_agency_plan(uid: str) -> dict:
uid = str(uid or "").strip()
if not uid:
return {}
data = _load_agency_plans()
plan = data.get("plans", {}).get(uid, {})
return plan if isinstance(plan, dict) else {}
def agency_plan_public(uid: str) -> dict:
plan = get_agency_plan(uid)
monthly = max(0, _agency_int(plan.get("monthly_credits"), 0))
used = max(0, _agency_int(plan.get("used_credits"), 0))
remaining = max(0, monthly - used)
return {
"active": bool(plan.get("active", False)),
"plan_name": str(plan.get("plan_name", "Agenzia Creativa") or "Agenzia Creativa"),
"monthly_credits": monthly,
"used_credits": used,
"remaining_credits": remaining,
"renewal_date": str(plan.get("renewal_date", "") or ""),
"updated_at": str(plan.get("updated_at", "") or ""),
"notes": str(plan.get("notes", "") or "")
}
def agency_debit_or_none(uid: str, op: str, request_id: str = ""):
cost = agency_cert_cost(op)
if cost <= 0:
return False, None
uid = str(uid or "").strip()
data = _load_agency_plans()
plans = data.setdefault("plans", {})
plan = plans.get(uid, {})
if not isinstance(plan, dict) or not bool(plan.get("active", False)):
return False, None
monthly = max(0, _agency_int(plan.get("monthly_credits"), 0))
used_before = max(0, _agency_int(plan.get("used_credits"), 0))
remaining_before = max(0, monthly - used_before)
if remaining_before < cost:
return False, {
"reason": "insufficient_agency_cert_credits",
"cost": cost,
"remaining_before": remaining_before,
"monthly_credits": monthly,
"used_before": used_before
}
used_after = used_before + cost
now = _now_iso()
plan.update({
"active": bool(plan.get("active", True)),
"plan_name": str(plan.get("plan_name", "Agenzia Creativa") or "Agenzia Creativa"),
"monthly_credits": monthly,
"used_credits": used_after,
"updated_at": now
})
plans[uid] = plan
_save_agency_plans(data)
row = {
"ts": now,
"kind": "agency_cert_debit",
"user_id": uid,
"op": op,
"request_id": request_id,
"credit_cost": cost,
"used_before": used_before,
"used_after": used_after,
"remaining_before": remaining_before,
"remaining_after": max(0, monthly - used_after)
}
try:
_append_jsonl(AGENCY_USAGE_LOG, row)
except Exception:
pass
return True, row
def agency_refund(uid: str, op: str, credit_cost: int, request_id: str = "", reason: str = "refund"):
credit_cost = max(0, _agency_int(credit_cost, 0))
if credit_cost <= 0:
return agency_plan_public(uid)
data = _load_agency_plans()
plans = data.setdefault("plans", {})
plan = plans.get(str(uid), {})
if not isinstance(plan, dict):
return {}
used_before = max(0, _agency_int(plan.get("used_credits"), 0))
used_after = max(0, used_before - credit_cost)
plan["used_credits"] = used_after
plan["updated_at"] = _now_iso()
plans[str(uid)] = plan
_save_agency_plans(data)
try:
_append_jsonl(AGENCY_USAGE_LOG, {
"ts": _now_iso(),
"kind": "agency_cert_refund",
"user_id": str(uid),
"op": op,
"request_id": request_id,
"credit_cost": credit_cost,
"used_before": used_before,
"used_after": used_after,
"reason": reason
})
except Exception:
pass
return agency_plan_public(uid)
# ================== PROMO BONUS (Stripe / WELCOM1) ==================
def ensure_promo_usage_file():
if not os.path.exists(PROMO_USAGE_FILE):
_json_save(PROMO_USAGE_FILE, {"codes": {}})
def _promo_norm_email(email):
return str(email or "").strip().lower()
def _promo_norm_code(code):
return str(code or "").strip().upper()
def promo_bonus_already_used(code, uid, email):
ensure_promo_usage_file()
data = _json_load(PROMO_USAGE_FILE, {"codes": {}})
code = _promo_norm_code(code)
uid = str(uid or "").strip()
email = _promo_norm_email(email)
item = (data.get("codes") or {}).get(code, {}) or {}
used_uids = item.get("uids", {}) or {}
used_emails = item.get("emails", {}) or {}
if uid and uid in used_uids:
return True
if email and email in used_emails:
return True
return False
def mark_promo_bonus_used(code, uid, email, payment_id, bonus_abo):
ensure_promo_usage_file()
data = _json_load(PROMO_USAGE_FILE, {"codes": {}})
data.setdefault("codes", {})
code = _promo_norm_code(code)
uid = str(uid or "").strip()
email = _promo_norm_email(email)
item = data["codes"].setdefault(code, {"uids": {}, "emails": {}, "payments": {}})
record = {
"user_id": uid,
"email": email,
"payment_id": str(payment_id or ""),
"bonus_abo": float(bonus_abo)
}
if uid:
item.setdefault("uids", {})[uid] = record
if email:
item.setdefault("emails", {})[email] = record
if payment_id:
item.setdefault("payments", {})[str(payment_id)] = record
_json_save(PROMO_USAGE_FILE, data)
def _stripe_obj_to_dict(obj):
try:
return obj.to_dict_recursive()
except Exception:
try:
return dict(obj)
except Exception:
return obj if isinstance(obj, dict) else {}
def _stripe_get_promo_code_from_discount(discount):
discount = _stripe_obj_to_dict(discount) if not isinstance(discount, dict) else discount
promo = discount.get("promotion_code")
if isinstance(promo, dict):
code = promo.get("code")
if code:
return _promo_norm_code(code)
if isinstance(promo, str) and promo.startswith("promo_"):
try:
pc = stripe.PromotionCode.retrieve(promo)
pc = _stripe_obj_to_dict(pc)
code = pc.get("code")
if code:
return _promo_norm_code(code)
except Exception as e:
print("[STRIPE PROMO DEBUG] promotion_code retrieve failed:", str(e), flush=True)
source = discount.get("source") or {}
if isinstance(source, dict):
coupon = source.get("coupon")
if isinstance(coupon, dict):
name = str(coupon.get("name") or "")
cid = str(coupon.get("id") or "")
if WELCOME_PROMO_CODE in name.upper() or WELCOME_PROMO_CODE in cid.upper():
return WELCOME_PROMO_CODE
coupon = discount.get("coupon")
if isinstance(coupon, dict):
name = str(coupon.get("name") or "")
cid = str(coupon.get("id") or "")
if WELCOME_PROMO_CODE in name.upper() or WELCOME_PROMO_CODE in cid.upper():
return WELCOME_PROMO_CODE
return ""
def stripe_session_promo_codes(session_id, fallback_obj=None):
codes = set()
session_obj = fallback_obj or {}
try:
session_obj = stripe.checkout.Session.retrieve(
session_id,
expand=[
"total_details.breakdown.discounts.discount",
"total_details.breakdown.discounts.discount.promotion_code",
"total_details.breakdown.discounts.discount.source.coupon"
]
)
except Exception as e:
print("[STRIPE PROMO DEBUG] expanded session retrieve failed:", str(e), flush=True)
try:
session_obj = stripe.checkout.Session.retrieve(session_id)
except Exception as e2:
print("[STRIPE PROMO DEBUG] session retrieve failed:", str(e2), flush=True)
session_obj = fallback_obj or {}
session_obj = _stripe_obj_to_dict(session_obj)
discount_lists = []
total_details = session_obj.get("total_details") or {}
if isinstance(total_details, dict):
breakdown = total_details.get("breakdown") or {}
if isinstance(breakdown, dict):
discount_lists.append(breakdown.get("discounts") or [])
discount_lists.append(session_obj.get("discounts") or [])
for dlst in discount_lists:
if not isinstance(dlst, list):
continue
for item in dlst:
item = _stripe_obj_to_dict(item) if not isinstance(item, dict) else item
discount = item.get("discount", item)
if isinstance(discount, str):
continue
code = _stripe_get_promo_code_from_discount(discount)
if code:
codes.add(code)
return sorted(codes), session_obj
def stripe_session_email(session_obj, fallback_obj=None):
session_obj = session_obj or {}
fallback_obj = fallback_obj or {}
def pick(obj):
if not isinstance(obj, dict):
return ""
details = obj.get("customer_details") or {}
if isinstance(details, dict) and details.get("email"):
return details.get("email")
if obj.get("customer_email"):
return obj.get("customer_email")
return ""
return _promo_norm_email(pick(session_obj) or pick(fallback_obj))
# ================== PRICING (ABO per operazione) ==================
def ensure_pricing_file():
if not os.path.exists(PRICING_FILE):
_json_save(PRICING_FILE, {"default_op_cost_abo": 1.0, "operations": {"generic": 1.0}})
def get_pricing():
ensure_pricing_file()
data = _json_load(PRICING_FILE, {"default_op_cost_abo": 1.0, "operations": {"generic": 1.0}})
data.setdefault("default_op_cost_abo", 1.0)
data.setdefault("operations", {"generic": data["default_op_cost_abo"]})
return data
def op_cost_abo(op: str) -> float:
p = get_pricing()
op = (op or "generic").strip().lower()
ops = p.get("operations", {}) or {}
if op in ops:
try:
return float(ops[op])
except Exception:
return float(p.get("default_op_cost_abo", 1.0))
return float(p.get("default_op_cost_abo", 1.0))
def debit_or_reject(uid: str, op: str):
cost = op_cost_abo(op)
bal = get_balance(uid)
if bal < cost:
return False, bal, cost
newb = add_balance(uid, -cost)
return True, newb, cost
# ================== USAGE (log) ==================
def cert_find_seen(uid: str, request_id: str):
"""
Ritorna dict log se già visto (con eventuale certificate_path), altrimenti None.
"""
request_id = (request_id or "").strip()
if not request_id:
return None
try:
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
if str(row.get("user_id")) == str(uid) and str(row.get("request_id")) == request_id:
return row
except Exception:
continue
except Exception:
return None
return None
def cert_log_request(**kw):
_append_jsonl(CERT_USAGE_LOG, {"ts": _now_iso(), **kw})
def log_ai_usage(**kw):
_append_jsonl(USAGE_LOG, {"ts": _now_iso(), **kw})
def summarize_usage(days=30):
since = time.time() - int(days) * 86400
res = {"by_user": {}, "total_requests": 0, "total_ok": 0, "total_cost_eur": 0.0}
if not os.path.exists(USAGE_LOG):
return {**res, "days": int(days), "from": datetime.fromtimestamp(since, timezone.utc).isoformat(), "to": _now_iso()}
with open(USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
ts = datetime.fromisoformat(row["ts"]).timestamp()
if ts < since:
continue
uid = str(row.get("user_id", "unknown"))
u = res["by_user"].setdefault(uid, {"requests": 0, "ok": 0, "cost_eur": 0.0, "providers": {}})
u["requests"] += 1
res["total_requests"] += 1
if row.get("ok"):
u["ok"] += 1
res["total_ok"] += 1
cost = float(row.get("cost_eur", 0.0) or 0.0)
u["cost_eur"] += cost
res["total_cost_eur"] += cost
p = row.get("provider", "unknown")
u["providers"][p] = u["providers"].get(p, 0) + 1
except Exception:
pass
return {**res, "days": int(days), "from": datetime.fromtimestamp(since, timezone.utc).isoformat(), "to": _now_iso()}
# ================== PDF CERTIFICATE ==================
def _qr_to_temp_png(data: str, out_dir: str, filename: str) -> str:
if not HAS_QRCODE:
return ""
try:
img = qrcode.make(data)
path = os.path.join(out_dir, filename)
img.save(path)
return path
except Exception:
return ""
def generate_base_certificate(payload: dict, uid: str, request_id: str) -> str:
"""
Crea PDF base con campi:
cert_id, work_id, deposit_date, author, original_filename, verify_url, file_hash
"""
os.makedirs(CERT_BASE_DIR, exist_ok=True)
cert_id = str(payload.get("cert_id", "")).strip()
work_id = str(payload.get("work_id", "")).strip()
deposit_date = str(payload.get("deposit_date", "")).strip()
author = str(payload.get("author", "")).strip()
original_filename = str(payload.get("original_filename", "")).strip()
verify_url = str(payload.get("verify_url", "")).strip()
file_hash = str(payload.get("file_hash", "")).strip()
video_proof = bool(payload.get("video_proof", False) or payload.get("video_sha256", "") or payload.get("video_hash", "") or payload.get("video_declaration_text", ""))
video_sha256 = str(payload.get("video_sha256", "") or payload.get("video_hash", "") or "").strip()
video_declaration_text = str(payload.get("video_declaration_text", "") or "").strip()
video_file_name = str(payload.get("video_file_name", "") or "").strip()
video_retained_by_platform = bool(payload.get("video_retained_by_platform", False))
content_type = str(payload.get("content_type", "") or "").strip()
ai_tool_used = str(payload.get("ai_tool_used", "") or "").strip()
ai_prompt = str(payload.get("ai_prompt", "") or "").strip()
human_contribution = str(payload.get("human_contribution", "") or "").strip()
ai_process_notes = str(payload.get("ai_process_notes", "") or "").strip()
ai_declaration_accepted = bool(payload.get("ai_declaration_accepted", False))
ai_assisted = bool(content_type == "ai_assisted" or ai_tool_used or ai_prompt or human_contribution or ai_process_notes or ai_declaration_accepted)
filename = f"cert_base_{uid}_{request_id}.pdf"
fullpath = os.path.join(CERT_BASE_DIR, filename)
# QR code temp
qr_path = ""
if verify_url:
qr_path = _qr_to_temp_png(verify_url, CERT_BASE_DIR, f"qr_{uid}_{request_id}.png")
c = canvas.Canvas(fullpath, pagesize=A4)
w, h = A4
# Header
c.setFont("Helvetica-Bold", 20)
c.drawString(2*cm, h - 2.5*cm, "COPYRIGHTCHAIN")
c.setFont("Helvetica", 11)
c.drawString(2*cm, h - 3.2*cm, "Certificato di Copyright - Deposito con Data Certa (BASE)")
# Box dati
y = h - 5.0*cm
c.setFont("Helvetica-Bold", 12)
c.drawString(2*cm, y, "Dati Certificato")
y -= 0.8*cm
c.setFont("Helvetica", 11)
def line(label, value):
nonlocal y
if value:
c.setFont("Helvetica-Bold", 10)
c.drawString(2*cm, y, f"{label}:")
c.setFont("Helvetica", 10)
c.drawString(6.2*cm, y, str(value))
y -= 0.6*cm
def multiline(label, value, max_len=80, max_lines=4):
nonlocal y
value = str(value or "").strip()
if not value:
return
chunks = [value[i:i+max_len] for i in range(0, len(value), max_len)]
if len(chunks) > max_lines:
chunks = chunks[:max_lines]
chunks[-1] = chunks[-1][:max(0, max_len-6)] + " [...]"
for i, chunk in enumerate(chunks):
if i == 0:
c.setFont("Helvetica-Bold", 10)
c.drawString(2*cm, y, f"{label}:")
c.setFont("Helvetica", 10)
c.drawString(6.2*cm, y, chunk)
y -= 0.5*cm
line("Cert ID", cert_id or f"{uid}-{request_id}")
line("Work ID", work_id)
line("Deposit Date", deposit_date)
line("Author / Owner", author or uid)
line("Original filename", original_filename)
line("File hash (SHA256)", file_hash)
if video_proof:
line("Video Proof", "Presente - dichiarazione video associata al deposito")
multiline("Video hash SHA256", video_sha256, max_len=80, max_lines=2)
line("Video file", video_file_name or "file video scaricato dall'utente")
line("Video conservato", "No" if not video_retained_by_platform else "Si")
multiline("Dichiarazione", video_declaration_text, max_len=80, max_lines=4)
if ai_assisted:
line("AI Proof", "Contenuto creato o modificato con AI")
line("Strumento AI", ai_tool_used)
multiline("Prompt principale", ai_prompt, max_len=80, max_lines=3)
multiline("Contributo umano", human_contribution, max_len=80, max_lines=5)
multiline("Note processo", ai_process_notes, max_len=80, max_lines=3)
line("Dichiarazione AI", "Accettata" if ai_declaration_accepted else "Non indicata")
if verify_url:
line("Verify URL", verify_url)
# QR
if qr_path and os.path.exists(qr_path):
try:
c.drawImage(qr_path, w - 6.5*cm, h - 7.5*cm, width=4.5*cm, height=4.5*cm, preserveAspectRatio=True, mask='auto')
c.setFont("Helvetica", 8)
c.drawRightString(w - 2*cm, h - 7.8*cm, "Scansiona per verificare")
except Exception:
pass
# Footer
c.setFont("Helvetica-Oblique", 9)
c.drawString(2*cm, 2.0*cm, f"Generated at (UTC): {datetime.now(timezone.utc).isoformat()}")
c.setFont("Helvetica", 8)
c.drawString(2*cm, 1.4*cm, "Questo certificato attesta il deposito del contenuto e l'associazione univoca ai dati sopra indicati.")
c.showPage()
c.save()
# cleanup QR temp (non obbligatorio)
try:
if qr_path and os.path.exists(qr_path):
os.remove(qr_path)
except Exception:
pass
return fullpath
# ================== CERT CORE ==================
def _openapi_timestamp_for_hash(file_hash: str) -> dict:
"""
Crea una marca temporale Openapi/InfoCert sull'hash SHA-256 del file.
Il file originale non viene inviato a Openapi.
"""
file_hash = str(file_hash or "").strip().lower()
if not re.fullmatch(r"[0-9a-f]{64}", file_hash):
raise ValueError("invalid_or_missing_file_hash")
if not OPENAPI_TIMESTAMP_ENABLED:
raise RuntimeError("openapi_timestamp_disabled")
if not OPENAPI_TIMESTAMP_TOKEN or not OPENAPI_TIMESTAMP_LOTTO_USERNAME or not OPENAPI_TIMESTAMP_LOTTO_PASSWORD:
raise RuntimeError("openapi_timestamp_missing_credentials")
base_url = "https://test.ws.marchetemporali.com" if OPENAPI_TIMESTAMP_ENV == "sandbox" else "https://ws.marchetemporali.com"
body = {
"username": OPENAPI_TIMESTAMP_LOTTO_USERNAME,
"password": OPENAPI_TIMESTAMP_LOTTO_PASSWORD,
"file": file_hash,
"mime": False,
"type": OPENAPI_TIMESTAMP_TYPE or "infocert",
}
r = requests.post(
base_url + "/marca",
headers={
"Authorization": f"Bearer {OPENAPI_TIMESTAMP_TOKEN}",
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "curl/8.0",
},
json=body,
timeout=OPENAPI_TIMESTAMP_TIMEOUT,
)
try:
res = r.json()
except Exception:
raise RuntimeError(f"openapi_timestamp_non_json_response_http_{r.status_code}")
data = res.get("data") or {}
ok = bool(res.get("success") is True)
meta = {
"status": "issued" if ok else "failed",
"provider": OPENAPI_TIMESTAMP_PROVIDER or "Openapi / InfoCert",
"environment": OPENAPI_TIMESTAMP_ENV,
"type": OPENAPI_TIMESTAMP_TYPE or "infocert",
"hash": file_hash,
"timestamp_issued_at": _now_iso(),
"transaction": str(data.get("transaction", "") or ""),
"timestamp_body_url": str(data.get("timestamp_body", "") or ""),
"timestamp_header": str(data.get("timestamp_header", "") or ""),
"available": data.get("available"),
"used": data.get("used"),
"message": str(res.get("message", "") or ""),
"error": res.get("error"),
"http_status": r.status_code,
}
if not ok:
raise RuntimeError(f"openapi_timestamp_failed: {meta.get('message') or meta.get('error') or r.status_code}")
return meta
def generate_timestamp_certificate(payload: dict, uid: str, request_id: str, timestamp_meta: dict) -> str:
"""
Genera PDF certificato Marca Temporale.
"""
os.makedirs(CERT_TS_DIR, exist_ok=True)
cert_id = str(payload.get("cert_id", "") or "").strip()
file_hash = str(payload.get("file_hash", "") or "").strip()
verify_url = str(payload.get("verify_url", "") or "").strip()
original_filename = str(payload.get("original_filename", "") or "").strip()
author = str(payload.get("author", "") or "").strip()
deposit_date = str(payload.get("deposit_date", "") or "").strip() or _now_iso()
header_raw = str(timestamp_meta.get("timestamp_header", "") or "")
header = {}
if header_raw:
try:
header = json.loads(header_raw)
except Exception:
header = {}
out = os.path.join(CERT_TS_DIR, f"cert_timestamp_{uid}_{request_id}.pdf")
c = canvas.Canvas(out, pagesize=A4)
w, h = A4
c.setFont("Helvetica-Bold", 18)
c.drawString(2*cm, h - 2.5*cm, "CERTIFICATO MARCA TEMPORALE")
c.setFont("Helvetica", 10)
y = h - 4.0*cm
def line(label, value, max_len=105):
nonlocal y
value = str(value or "").strip()
if not value:
return
c.drawString(2*cm, y, f"{label}: {value[:max_len]}")
y -= 0.62*cm
def multiline(label, value, max_len=82, max_lines=3):
nonlocal y
value = str(value or "").strip()
if not value:
return
chunks = [value[i:i+max_len] for i in range(0, len(value), max_len)]
if len(chunks) > max_lines:
chunks = chunks[:max_lines]
chunks[-1] = chunks[-1][:max(0, max_len-6)] + " [...]"
for i, chunk in enumerate(chunks):
c.drawString(2*cm, y, f"{label}: {chunk}" if i == 0 else f" {chunk}")
y -= 0.52*cm
line("User ID", uid)
line("Request ID", request_id)
line("Cert ID", cert_id)
line("Autore / Titolare", author)
line("Nome file", original_filename)
line("Data deposito", deposit_date)
y -= 0.25*cm
c.setFont("Helvetica-Bold", 12)
c.drawString(2*cm, y, "Oggetto della marcatura")
y -= 0.75*cm
c.setFont("Helvetica", 10)
multiline("Hash SHA-256 marcato", file_hash, max_len=90, max_lines=2)
line("File originale inviato a Openapi", "No")
line("Modalita", "Marcatura dell'impronta SHA-256 del file")
y -= 0.25*cm
c.setFont("Helvetica-Bold", 12)
c.drawString(2*cm, y, "Dati marca temporale")
y -= 0.75*cm
c.setFont("Helvetica", 10)
line("Provider", timestamp_meta.get("provider", ""))
line("Ambiente", timestamp_meta.get("environment", ""))
line("Tipo", timestamp_meta.get("type", ""))
line("Stato", timestamp_meta.get("status", ""))
line("Transaction", timestamp_meta.get("transaction", ""))
line("Issued at UTC", timestamp_meta.get("timestamp_issued_at", ""))
line("InfoCert UCTTime", header.get("ICTSA-UCTTime", ""))
line("InfoCert Serial Number", header.get("ICTSA-SN", ""))
multiline("InfoCert TSA", header.get("ICTSA-TSAName", ""), max_len=90, max_lines=2)
multiline("Timestamp URL", timestamp_meta.get("timestamp_body_url", ""), max_len=90, max_lines=2)
c.setFont("Helvetica-Oblique", 9)
c.drawString(2*cm, 2.2*cm, f"Verify URL: {verify_url}")
c.drawString(2*cm, 1.6*cm, f"Generated at UTC: {_now_iso()}")
c.showPage()
c.save()
return out
def _resolve_uid_from_request(payload):
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok) if tok else None
if user:
return str(user["id"]), user, None
uid = payload.get("user_id")
if uid:
return str(uid), None, None
return None, None, (jsonify({"error": "unauthorized"}), 401)
def _build_public_cert_url(fullpath: str) -> str:
if not PUBLIC_BASE_URL:
return ""
# esponiamo via endpoint /cert/file?path=... (più sotto)
return f"{PUBLIC_BASE_URL}/cert/file?path={fullpath}"
def _build_verify_url(cert_id: str, file_hash: str = "") -> str:
if not VERIFY_BASE_URL:
return ""
cert_id = str(cert_id or "").strip()
file_hash = str(file_hash or "").strip()
if cert_id and file_hash:
return f"{VERIFY_BASE_URL}/?id={cert_id}&hash={file_hash}"
if cert_id:
return f"{VERIFY_BASE_URL}/?id={cert_id}"
if file_hash:
return f"{VERIFY_BASE_URL}/?hash={file_hash}"
return VERIFY_BASE_URL
def _cert_debit_core(payload):
cert_type = str(payload.get("cert_type", "base")).strip().lower()
request_id = str(payload.get("request_id", "")).strip()
if cert_type not in ("base", "blockchain", "timestamp"):
return jsonify({"error": "invalid_cert_type", "allowed": ["base", "blockchain", "timestamp"]}), 400
if not request_id:
return jsonify({"error": "missing_request_id"}), 400
uid, user, err = _resolve_uid_from_request(payload)
if err:
return err
if cert_type == "blockchain":
op = "cert_blockchain"
elif cert_type == "timestamp":
op = "cert_timestamp"
else:
op = "cert_base"
# idempotenza robusta: se già visto MA file non esiste → rigenera senza addebito
seen = cert_find_seen(uid, request_id)
if seen:
seen_path = str(seen.get("certificate_path", "") or "")
expected = seen_path
if not expected:
# fallback: ricostruiamo path standard
if cert_type == "blockchain":
base_dir = CERT_BC_DIR
expected = os.path.join(base_dir, f"cert_blockchain_{uid}_{request_id}.pdf")
elif cert_type == "timestamp":
base_dir = CERT_TS_DIR
expected = os.path.join(base_dir, f"cert_timestamp_{uid}_{request_id}.pdf")
else:
base_dir = CERT_BASE_DIR
expected = os.path.join(base_dir, f"cert_base_{uid}_{request_id}.pdf")
if expected and os.path.exists(expected):
return jsonify({
"ok": True,
"user_id": uid,
"cert_type": cert_type,
"op": op,
"idempotent_replay": True,
"request_id": request_id,
"balance_abo": get_balance(uid),
"certificate_path": expected,
"certificate_url": _build_verify_url(seen.get("cert_id", ""), seen.get("file_hash", ""))
})
# se file manca: rigenera sotto senza addebito
# addebito: crediti certificazione Agenzia e wallet ABO restano separati.
# billing_wallet può essere:
# - "agency": usa solo crediti Agenzia per questa certificazione
# - "abo": usa solo ABO Wallet per questa certificazione
# - "auto"/vuoto: prova Agenzia se attiva, poi fallback ABO
requested_wallet = str(payload.get("billing_wallet", "auto") or "auto").strip().lower()
if requested_wallet not in ("agency", "abo", "auto"):
requested_wallet = "auto"
billing_method = "abo"
agency_credit_cost = 0
agency_debit = None
agency_attempted = False
agency_cost = agency_cert_cost(op)
if agency_cost > 0 and requested_wallet != "abo":
agency_attempted = True
agency_ok, agency_debit = agency_debit_or_none(uid, op, request_id)
if agency_ok:
billing_method = "agency_cert_credits"
agency_credit_cost = int(agency_debit.get("credit_cost") or agency_cost)
bal_before = get_balance(uid)
cost_abo = 0.0
elif requested_wallet == "agency":
return jsonify({
"error": "insufficient_agency_cert_credits",
"op": op,
"cert_type": cert_type,
"billing_wallet": requested_wallet,
"cost_abo": op_cost_abo(op),
"balance_abo": get_balance(uid),
"agency": agency_plan_public(uid),
"agency_required_credits": agency_cost,
"agency_attempt": agency_debit,
"message": "Crediti Agenzia insufficienti. Se vuoi usare gli ABO, seleziona ABO Wallet."
}), 402
if billing_method != "agency_cert_credits":
ok_debit, bal_before, cost_abo = debit_or_reject(uid, op)
if not ok_debit:
return jsonify({
"error": "insufficient_balance",
"op": op,
"cert_type": cert_type,
"billing_wallet": requested_wallet,
"cost_abo": cost_abo,
"balance_abo": bal_before,
"agency": agency_plan_public(uid),
"agency_required_credits": agency_cost,
"agency_attempt": agency_debit
}), 402
cert_id = f"CD-{abs(hash(f'{uid}-{request_id}')) % 1000000:06d}"
file_hash = str(payload.get("file_hash", "")).strip()
verify_url = _build_verify_url(cert_id, file_hash)
payload = dict(payload or {})
payload["verify_url"] = verify_url
payload["cert_id"] = cert_id
content_type = str(payload.get("content_type", "") or "").strip()
ai_tool_used = str(payload.get("ai_tool_used", "") or "").strip()
ai_prompt = str(payload.get("ai_prompt", "") or "").strip()
human_contribution = str(payload.get("human_contribution", "") or "").strip()
ai_process_notes = str(payload.get("ai_process_notes", "") or "").strip()
ai_declaration_accepted = bool(payload.get("ai_declaration_accepted", False))
ai_assisted = bool(content_type == "ai_assisted" or ai_tool_used or ai_prompt or human_contribution or ai_process_notes or ai_declaration_accepted)
ots_meta = {
"status": "pending",
"ots_path": "",
"ots_manifest_path": "",
"btc_txid": "",
"btc_block_height": None,
"ots_last_upgrade_at": "",
"anchored_at": "",
"confirmed_at": "",
}
timestamp_meta = {
"status": "",
"provider": "",
"environment": "",
"type": "",
"hash": "",
"timestamp_issued_at": "",
"transaction": "",
"timestamp_body_url": "",
"timestamp_header": "",
"available": None,
"used": None,
"message": "",
"error": None,
"http_status": None,
}
# genera PDF
try:
if cert_type == "base":
certificate_path = generate_base_certificate(payload, uid, request_id)
elif cert_type == "timestamp":
timestamp_meta = _openapi_timestamp_for_hash(file_hash)
payload["timestamp_meta"] = timestamp_meta
certificate_path = generate_timestamp_certificate(payload, uid, request_id, timestamp_meta)
else:
ots_meta = _ots_stamp_for_payload(payload, uid, request_id, cert_id, verify_url)
os.makedirs(CERT_BC_DIR, exist_ok=True)
certificate_path = os.path.join(CERT_BC_DIR, f"cert_blockchain_{uid}_{request_id}.pdf")
c = canvas.Canvas(certificate_path, pagesize=A4)
w, h = A4
c.setFont("Helvetica-Bold", 18)
c.drawString(2*cm, h - 3*cm, "CERTIFICATO BLOCKCHAIN (OpenTimestamps)")
c.setFont("Helvetica", 11)
yb = h - 4.2*cm
c.drawString(2*cm, yb, f"User ID: {uid}")
yb -= 0.8*cm
c.drawString(2*cm, yb, f"Request ID: {request_id}")
yb -= 0.8*cm
c.drawString(2*cm, yb, f"Cert ID: {cert_id}")
yb -= 0.8*cm
c.drawString(2*cm, yb, f"File Hash: {file_hash}")
yb -= 0.8*cm
def bc_multiline(label, value, max_len=72, max_lines=3):
nonlocal yb
value = str(value or "").strip()
if not value:
return
chunks = [value[i:i+max_len] for i in range(0, len(value), max_len)]
if len(chunks) > max_lines:
chunks = chunks[:max_lines]
chunks[-1] = chunks[-1][:max(0, max_len-6)] + " [...]"
for i, chunk in enumerate(chunks):
c.drawString(2*cm, yb, f"{label}: {chunk}" if i == 0 else f" {chunk}")
yb -= 0.55*cm
video_present = bool(payload.get("video_proof", False) or payload.get("video_sha256", "") or payload.get("video_hash", "") or payload.get("video_declaration_text", ""))
if video_present:
c.drawString(2*cm, yb, "Video Proof: Presente - dichiarazione video associata al deposito")
yb -= 0.8*cm
video_sha = str(payload.get("video_sha256", "") or payload.get("video_hash", "") or "").strip()
if video_sha:
c.drawString(2*cm, yb, f"Video SHA256: {video_sha}")
yb -= 0.8*cm
c.drawString(2*cm, yb, "Video conservato da CopyrightChain: No")
yb -= 0.8*cm
if ai_assisted:
c.setFont("Helvetica-Bold", 11)
c.drawString(2*cm, yb, "AI Proof: Contenuto creato o modificato con AI")
yb -= 0.8*cm
c.setFont("Helvetica", 10)
if ai_tool_used:
c.drawString(2*cm, yb, f"Strumento AI: {ai_tool_used}")
yb -= 0.6*cm
bc_multiline("Prompt principale", ai_prompt, max_len=72, max_lines=2)
bc_multiline("Contributo umano", human_contribution, max_len=72, max_lines=4)
bc_multiline("Note processo", ai_process_notes, max_len=72, max_lines=2)
c.drawString(2*cm, yb, f"Dichiarazione AI: {'Accettata' if ai_declaration_accepted else 'Non indicata'}")
yb -= 0.8*cm
c.setFont("Helvetica", 11)
c.drawString(2*cm, yb, f"OTS Status: {ots_meta.get('status','pending')}")
yb -= 0.8*cm
if ots_meta.get("btc_txid"):
c.drawString(2*cm, yb, f"Bitcoin TxID: {ots_meta.get('btc_txid','')}")
yb -= 0.8*cm
if ots_meta.get("btc_block_height") is not None:
c.drawString(2*cm, yb, f"Block Height: {ots_meta.get('btc_block_height')}")
c.setFont("Helvetica-Oblique", 9)
c.drawString(2*cm, 2.8*cm, f"OTS Proof: {ots_meta.get('ots_path','')}")
c.drawString(2*cm, 2.2*cm, f"Verify URL: {verify_url}")
c.drawString(2*cm, 1.6*cm, f"Generated at (UTC): {_now_iso()}")
c.showPage()
c.save()
except Exception as e:
# se fallisce la generazione: rimborsa il metodo usato
if billing_method == "agency_cert_credits" and agency_credit_cost > 0:
agency_refund(uid, op, agency_credit_cost, request_id, "certificate_generation_failed")
else:
add_balance(uid, cost_abo)
log_ai_usage(
user_id=uid, provider="gateway", ok=False, status=500, ms=0, cost_eur=0.0,
op=op, abo_cost=cost_abo, cert_type=cert_type, request_id=request_id,
billing_method=billing_method, agency_credit_cost=agency_credit_cost,
error="certificate_generation_failed"
)
return jsonify({"error": "certificate_generation_failed", "details": str(e)}), 500
# log certificato (idempotenza) SOLO dopo successo generazione
cert_log_request(
user_id=uid,
request_id=request_id,
op=op,
cert_type=cert_type,
abo_cost=cost_abo,
billing_method=billing_method,
agency_credit_cost=agency_credit_cost,
agency_debit=agency_debit,
certificate_path=certificate_path,
cert_id=cert_id,
file_hash=file_hash,
video_proof=bool(payload.get("video_proof", False) or payload.get("video_sha256", "") or payload.get("video_declaration_text", "")),
video_sha256=str(payload.get("video_sha256", "") or payload.get("video_hash", "") or "").strip(),
video_declaration_text=str(payload.get("video_declaration_text", "") or "").strip(),
video_file_name=str(payload.get("video_file_name", "") or "").strip(),
video_retained_by_platform=bool(payload.get("video_retained_by_platform", False)),
content_type=content_type,
ai_assisted=ai_assisted,
ai_tool_used=ai_tool_used,
ai_prompt=ai_prompt,
human_contribution=human_contribution,
ai_process_notes=ai_process_notes,
ai_declaration_accepted=ai_declaration_accepted,
verify_url=verify_url,
original_filename=str(payload.get("original_filename", "")).strip(),
author=str(payload.get("author", "")).strip(),
deposit_date=str(payload.get("deposit_date", "")).strip() or _now_iso(),
blockchain_status=ots_meta.get("status", "pending") if cert_type == "blockchain" else "",
ots_path=ots_meta.get("ots_path", ""),
ots_manifest_path=ots_meta.get("ots_manifest_path", ""),
ots_last_upgrade_at=ots_meta.get("ots_last_upgrade_at", ""),
btc_txid=ots_meta.get("btc_txid", ""),
btc_block_height=ots_meta.get("btc_block_height"),
anchored_at=ots_meta.get("anchored_at", ""),
confirmed_at=ots_meta.get("confirmed_at", ""),
timestamp_status=timestamp_meta.get("status", ""),
timestamp_provider=timestamp_meta.get("provider", ""),
timestamp_environment=timestamp_meta.get("environment", ""),
timestamp_type=timestamp_meta.get("type", ""),
timestamp_hash=timestamp_meta.get("hash", ""),
timestamp_issued_at=timestamp_meta.get("timestamp_issued_at", ""),
timestamp_transaction=timestamp_meta.get("transaction", ""),
timestamp_body_url=timestamp_meta.get("timestamp_body_url", ""),
timestamp_header=timestamp_meta.get("timestamp_header", ""),
timestamp_available=timestamp_meta.get("available"),
timestamp_used=timestamp_meta.get("used"),
timestamp_message=timestamp_meta.get("message", ""),
timestamp_http_status=timestamp_meta.get("http_status")
)
# log generale
log_ai_usage(
user_id=uid, provider="gateway", ok=True, status=200, ms=0, cost_eur=0.0,
op=op, abo_cost=cost_abo, cert_type=cert_type, request_id=request_id,
billing_method=billing_method, agency_credit_cost=agency_credit_cost
)
return jsonify({
"ok": True,
"user_id": uid,
"cert_type": cert_type,
"op": op,
"billing_method": billing_method,
"billing_wallet": requested_wallet,
"debited_abo": cost_abo,
"agency_credit_cost": agency_credit_cost,
"agency": agency_plan_public(uid),
"idempotent_replay": False,
"request_id": request_id,
"balance_abo": get_balance(uid),
"certificate_path": certificate_path,
"certificate_url": verify_url,
"verify_url": verify_url,
"cert_id": cert_id,
"file_hash": file_hash,
"video_proof": bool(payload.get("video_proof", False) or payload.get("video_sha256", "") or payload.get("video_declaration_text", "")),
"video_sha256": str(payload.get("video_sha256", "") or payload.get("video_hash", "") or "").strip(),
"video_retained_by_platform": bool(payload.get("video_retained_by_platform", False)),
"content_type": content_type,
"ai_assisted": ai_assisted,
"ai_tool_used": ai_tool_used,
"ai_prompt": ai_prompt,
"human_contribution": human_contribution,
"ai_process_notes": ai_process_notes,
"ai_declaration_accepted": ai_declaration_accepted,
"blockchain_status": ots_meta.get("status", "") if cert_type == "blockchain" else "",
"ots": ots_meta if cert_type == "blockchain" else None,
"timestamp_status": timestamp_meta.get("status", "") if cert_type == "timestamp" else "",
"timestamp": timestamp_meta if cert_type == "timestamp" else None
})
def _safe_fs(value: str) -> str:
return re.sub(r"[^A-Za-z0-9._-]+", "_", str(value or "").strip())[:180]
def _run_ots(*args, timeout=180):
env = dict(os.environ)
env["PATH"] = f"{os.path.dirname(OTS_BIN)}:{env.get('PATH','')}"
proc = subprocess.run(
[OTS_BIN, *list(args)],
capture_output=True,
text=True,
timeout=timeout,
env=env,
)
out = ((proc.stdout or "") + "\n" + (proc.stderr or "")).strip()
return proc.returncode, out
def _parse_ots_status(text: str):
text = text or ""
txids = re.findall(r"(?:Transaction id|Timestamped by transaction)\s+([0-9a-fA-F]{64})", text)
heights = [int(x) for x in re.findall(r"BitcoinBlockHeaderAttestation\((\d+)\)", text)]
if not heights:
m = re.search(r"block(?:\s+height)?\s+(\d+)", text, re.I)
if m:
try:
heights = [int(m.group(1))]
except Exception:
heights = []
status = "pending"
if heights or re.search(r"Success! Timestamp complete|attests data existed as of", text, re.I):
status = "confirmed"
elif txids or re.search(r"waiting for \d+ confirmations", text, re.I):
status = "anchored"
elif re.search(r"PendingAttestation|Pending confirmation in Bitcoin blockchain", text, re.I):
status = "pending"
return {
"status": status,
"btc_txid": txids[-1] if txids else "",
"btc_txids": txids,
"btc_block_height": max(heights) if heights else None,
"btc_block_heights": heights,
"ots_raw": text[-12000:],
}
def _build_ots_manifest(payload: dict, uid: str, request_id: str, cert_id: str, verify_url: str):
file_hash = str(payload.get("file_hash", "")).strip()
if not file_hash:
raise ValueError("missing_file_hash")
return {
"schema": "copyrightchain-ots-v1",
"cert_id": str(cert_id),
"user_id": str(uid),
"request_id": str(request_id),
"file_hash": file_hash,
"hash_algorithm": "sha256",
"original_filename": str(payload.get("original_filename", "")).strip(),
"author": str(payload.get("author", "")).strip(),
"deposit_date": str(payload.get("deposit_date", "")).strip() or _now_iso(),
"content_type": str(payload.get("content_type", "") or "").strip(),
"ai_proof": {
"ai_assisted": bool(str(payload.get("content_type", "") or "").strip() == "ai_assisted" or payload.get("ai_tool_used") or payload.get("ai_prompt") or payload.get("human_contribution") or payload.get("ai_process_notes") or payload.get("ai_declaration_accepted")),
"ai_tool_used": str(payload.get("ai_tool_used", "") or "").strip(),
"ai_prompt": str(payload.get("ai_prompt", "") or "").strip(),
"human_contribution": str(payload.get("human_contribution", "") or "").strip(),
"ai_process_notes": str(payload.get("ai_process_notes", "") or "").strip(),
"ai_declaration_accepted": bool(payload.get("ai_declaration_accepted", False)),
},
"verify_url": str(verify_url or "").strip(),
"generated_at": _now_iso(),
}
def _rewrite_cert_usage_row(user_id: str, request_id: str, updates: dict):
if not os.path.exists(CERT_USAGE_LOG):
return False
changed = False
rows = []
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
except Exception:
continue
if str(row.get("user_id")) == str(user_id) and str(row.get("request_id")) == str(request_id):
row.update(updates or {})
changed = True
rows.append(row)
if changed:
tmp = CERT_USAGE_LOG + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
for row in rows:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
os.replace(tmp, CERT_USAGE_LOG)
return changed
def _ots_refresh_files(manifest_path: str, ots_path: str):
outputs = []
for cmd, timeout in (("upgrade", 90), ("verify", 90), ("info", 30)):
try:
rc, out = _run_ots(cmd, ots_path, timeout=timeout)
if out:
outputs.append(f"$ ots {cmd}\n{out}")
except Exception as e:
outputs.append(f"$ ots {cmd}\n{e}")
combined = "\n\n".join(outputs).strip()
meta = _parse_ots_status(combined)
meta["ots_last_upgrade_at"] = _now_iso()
return meta
def _ots_stamp_for_payload(payload: dict, uid: str, request_id: str, cert_id: str, verify_url: str):
manifest = _build_ots_manifest(payload, uid, request_id, cert_id, verify_url)
base = f"{_safe_fs(uid)}_{_safe_fs(request_id)}"
manifest_path = os.path.join(OTS_WORK_DIR, f"manifest_{base}.json")
ots_path = manifest_path + ".ots"
with open(manifest_path, "w", encoding="utf-8") as f:
json.dump(manifest, f, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
if os.path.exists(ots_path):
os.remove(ots_path)
rc, out = _run_ots("stamp", manifest_path, timeout=60)
if rc != 0 or not os.path.exists(ots_path):
raise RuntimeError(f"ots_stamp_failed: {out}")
meta = {
"status": "pending",
"ots_path": ots_path,
"ots_manifest_path": manifest_path,
"ots_stamped_at": _now_iso(),
"ots_last_upgrade_at": "",
"btc_txid": "",
"btc_block_height": None,
"anchored_at": "",
"confirmed_at": "",
}
refreshed = _ots_refresh_files(manifest_path, ots_path)
meta.update({
"status": refreshed.get("status", "pending"),
"ots_last_upgrade_at": refreshed.get("ots_last_upgrade_at", ""),
"btc_txid": refreshed.get("btc_txid", ""),
"btc_block_height": refreshed.get("btc_block_height"),
})
if meta["status"] in ("anchored", "confirmed"):
meta["anchored_at"] = _now_iso()
if meta["status"] == "confirmed":
meta["confirmed_at"] = _now_iso()
return meta
def _ots_refresh_row(row: dict):
row = dict(row or {})
if str(row.get("cert_type", "base")).strip() != "blockchain":
return row
manifest_path = str(row.get("ots_manifest_path", "") or "").strip()
ots_path = str(row.get("ots_path", "") or "").strip()
if not manifest_path or not ots_path or not os.path.exists(ots_path):
row["blockchain_status"] = str(row.get("blockchain_status", "") or "").strip() or "pending"
return row
refreshed = _ots_refresh_files(manifest_path, ots_path)
row["blockchain_status"] = refreshed.get("status", row.get("blockchain_status", "pending"))
row["ots_last_upgrade_at"] = refreshed.get("ots_last_upgrade_at", "")
row["btc_txid"] = refreshed.get("btc_txid") or row.get("btc_txid", "")
row["btc_block_height"] = refreshed.get("btc_block_height") if refreshed.get("btc_block_height") is not None else row.get("btc_block_height")
if row["blockchain_status"] in ("anchored", "confirmed") and not row.get("anchored_at"):
row["anchored_at"] = _now_iso()
if row["blockchain_status"] == "confirmed" and not row.get("confirmed_at"):
row["confirmed_at"] = _now_iso()
_rewrite_cert_usage_row(
str(row.get("user_id", "") or "").strip(),
str(row.get("request_id", "") or "").strip(),
{
"blockchain_status": row.get("blockchain_status", ""),
"ots_last_upgrade_at": row.get("ots_last_upgrade_at", ""),
"btc_txid": row.get("btc_txid", ""),
"btc_block_height": row.get("btc_block_height"),
"anchored_at": row.get("anchored_at", ""),
"confirmed_at": row.get("confirmed_at", ""),
}
)
return row
# ================== ROUTES ==================
@APP.get("/health")
def health():
return jsonify({
"ok": True,
"has_openai_key": bool(OPENAI_API_KEY),
"has_gemini_key": bool(GEMINI_API_KEY),
"has_qrcode": HAS_QRCODE,
"tower_configured": bool(TOWER_PASSWORD and SESSION_SECRET)
})
@APP.get("/openapi.json")
def openapi_json():
return jsonify({
"openapi": "3.0.0",
"info": {"title": "CCP AI Gateway", "version": "1.0.0"},
"paths": {
"/health": {"get": {"summary": "Health"}},
"/openapi.json": {"get": {"summary": "OpenAPI spec"}},
"/docs": {"get": {"summary": "HTML docs"}},
"/tower/login": {"post": {"summary": "Tower login"}},
"/tower/users/upsert": {"post": {"summary": "Upsert user"}},
"/tower/users/list": {"get": {"summary": "List users"}},
"/tower/ai/stats": {"get": {"summary": "AI stats"}},
"/tower/balances": {"get": {"summary": "Balances dump"}},
"/tower/balances/topup": {"post": {"summary": "Topup"}},
"/tower/balances/set": {"post": {"summary": "Set balance"}},
"/tower/pricing": {"get": {"summary": "Get pricing"}},
"/tower/pricing/set": {"post": {"summary": "Set pricing"}},
"/me/balance": {"get": {"summary": "My balance"}},
"/certify": {"post": {"summary": "Create & debit certificate (base/blockchain)"}},
"/cert/debit": {"post": {"summary": "Debit certificate (compat)"}},
"/cert/file": {"get": {"summary": "Download certificate by server path (needs tower or admin token)"}},
"/text": {"post": {"summary": "AI text"}}
}
})
@APP.get("/docs")
def docs():
html = """<!doctype html><html><head><meta charset="utf-8"><title>CCP AI Gateway Docs</title></head>
<body style="font-family:Arial,sans-serif;padding:20px">
<h2>CCP AI Gateway</h2>
<ul>
<li><a href="/health">/health</a></li>
<li><a href="/openapi.json">/openapi.json</a></li>
</ul>
<p>Endpoints principali: /tower/*, /certify, /text</p>
</body></html>"""
return make_response(html, 200, {"Content-Type": "text/html; charset=utf-8"})
@APP.post("/tower/login")
def tower_login():
data = request.get_json(silent=True) or {}
if data.get("password") != TOWER_PASSWORD:
return jsonify({"error": "unauthorized"}), 401
tok = _sign({"role": "tower_admin", "exp": time.time() + 86400})
resp = make_response(jsonify({"ok": True}))
# su localhost http: non mettere Secure, altrimenti curl cookie-jar ok ma browser no.
resp.set_cookie("ccp_tower", tok, httponly=True, secure=False, samesite="Lax", path="/")
return resp
@APP.post("/tower/users/upsert")
def tower_users_upsert():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
if not data.get("id"):
return jsonify({"error": "missing_id"}), 400
user = {
"id": str(data["id"]),
"token": data.get("token", ""),
"email": data.get("email", ""),
"name": data.get("name", ""),
"wallet": data.get("wallet", ""),
"plan": data.get("plan", "free"),
"ai_limit_monthly": int(data.get("ai_limit_monthly", 0) or 0),
"updated_at": _now_iso()
}
upsert_user(user)
return jsonify({"ok": True})
@APP.post("/tower/users/create")
def tower_users_create():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
uid = str(data.get("id", "")).strip()
email = str(data.get("email", "")).strip().lower()
name = str(data.get("name", "")).strip()
wallet = str(data.get("wallet", "")).strip()
plan = str(data.get("plan", "free") or "free").strip().lower()
token = str(data.get("token", "")).strip()
temp_password = str(data.get("temp_password", "")).strip()
initial_balance = data.get("initial_balance", 0)
if not uid:
return jsonify({"error": "missing_id"}), 400
if not email:
return jsonify({"error": "missing_email"}), 400
if not temp_password or len(temp_password) < 8:
return jsonify({"error": "weak_temp_password"}), 400
existing_by_email = find_user_by_email(email)
if existing_by_email and str(existing_by_email.get("id", "")) != uid:
return jsonify({"error": "email_exists"}), 409
ensure_users_file()
current = _json_load(USERS_FILE, {"users": []})
for u in current.get("users", []):
if str(u.get("id", "")).strip() == uid:
return jsonify({"error": "id_exists"}), 409
now = _now_iso()
user = {
"id": uid,
"token": token or secrets.token_hex(24),
"email": email,
"name": name,
"wallet": wallet,
"plan": plan or "free",
"ai_limit_monthly": int(data.get("ai_limit_monthly", 0) or 0),
"password_hash": _pw_hash(temp_password),
"must_change_password": True,
"created_at": now,
"updated_at": now
}
upsert_user(user)
try:
initial_balance = float(initial_balance or 0)
except Exception:
initial_balance = 0.0
set_balance(uid, initial_balance)
return jsonify({
"ok": True,
"user": _user_public(user),
"temp_password_set": True,
"initial_balance": initial_balance
})
@APP.post("/tower/users/reset_password")
def tower_users_reset_password():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "")).strip()
temp_password = str(data.get("temp_password", "")).strip()
if not uid:
return jsonify({"error": "missing_user_id"}), 400
ensure_users_file()
users_data = _json_load(USERS_FILE, {"users": []})
users = users_data.get("users", [])
target = None
for u in users:
if str(u.get("id", "")).strip() == uid:
target = u
break
if not target:
return jsonify({"error": "user_not_found"}), 404
if not temp_password or len(temp_password) < 8:
temp_password = "Temp" + secrets.token_hex(4) + "!"
target["password_hash"] = _pw_hash(temp_password)
target["must_change_password"] = True
target["updated_at"] = _now_iso()
upsert_user(target)
return jsonify({
"ok": True,
"user_id": uid,
"temp_password": temp_password,
"must_change_password": True
})
@APP.post("/tower/users/delete")
def tower_users_delete():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "")).strip()
if not uid:
return jsonify({"error": "missing_user_id"}), 400
ensure_users_file()
users_data = _json_load(USERS_FILE, {"users": []})
users = users_data.get("users", [])
target = None
kept = []
for u in users:
if str(u.get("id", "")).strip() == uid:
target = u
else:
kept.append(u)
if not target:
return jsonify({"error": "user_not_found"}), 404
if str(target.get("plan", "")).strip().lower() == "admin":
return jsonify({"error": "protected_admin_user"}), 403
users_data["users"] = kept
_json_save(USERS_FILE, users_data)
ensure_balances_file()
bal = _json_load(BALANCES_FILE, {"balances": {}})
bal.setdefault("balances", {}).pop(uid, None)
_json_save(BALANCES_FILE, bal)
return jsonify({
"ok": True,
"deleted_user_id": uid,
"deleted_email": target.get("email", ""),
"deleted_name": target.get("name", "")
})
@APP.get("/tower/users/list")
def tower_users_list():
g = _tower_guard()
if g:
return g
ensure_users_file()
return jsonify(_json_load(USERS_FILE, {"users": []}))
@APP.get("/tower/ai/stats")
def tower_ai_stats():
g = _tower_guard()
if g:
return g
days = int(request.args.get("days", "30") or "30")
days = max(1, min(days, 365))
return jsonify(summarize_usage(days))
@APP.get("/tower/users/spend_summary")
def tower_users_spend_summary():
g = _tower_guard()
if g:
return g
out = {}
def ensure_uid(uid):
uid = str(uid or "").strip()
if not uid:
return None
if uid not in out:
out[uid] = {
"user_id": uid,
"abo_spent_total": 0.0,
"ai_abo_spent": 0.0,
"cert_abo_spent": 0.0,
"ai_calls": 0,
"cert_ops": 0,
"total_ops": 0,
"last_ts": ""
}
return out[uid]
for path, bucket in [(USAGE_LOG, "ai"), (CERT_USAGE_LOG, "cert")]:
if not os.path.exists(path):
continue
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except Exception:
continue
uid = str(row.get("user_id", "")).strip()
item = ensure_uid(uid)
if not item:
continue
abo = row.get("abo_cost", 0) or 0
try:
abo = float(abo)
except Exception:
abo = 0.0
item["abo_spent_total"] += abo
item["total_ops"] += 1
ts = str(row.get("ts", "") or "")
if ts and (not item["last_ts"] or ts > item["last_ts"]):
item["last_ts"] = ts
if bucket == "ai":
item["ai_abo_spent"] += abo
item["ai_calls"] += 1
else:
item["cert_abo_spent"] += abo
item["cert_ops"] += 1
except Exception:
continue
return jsonify({"users": list(out.values())})
@APP.get("/tower/balances")
def tower_balances():
g = _tower_guard()
if g:
return g
ensure_balances_file()
return jsonify(_json_load(BALANCES_FILE, {"balances": {}}))
@APP.post("/tower/balances/topup")
def tower_balances_topup():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "")).strip()
amount = data.get("amount", None)
if not uid or amount is None:
return jsonify({"error": "missing_params"}), 400
try:
amount = float(amount)
except Exception:
return jsonify({"error": "invalid_amount"}), 400
newb = add_balance(uid, amount)
return jsonify({"ok": True, "user_id": uid, "balance_abo": newb})
@APP.post("/tower/balances/set")
def tower_balances_set():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "")).strip()
balance = data.get("balance", None)
if not uid or balance is None:
return jsonify({"error": "missing_params"}), 400
try:
balance = float(balance)
except Exception:
return jsonify({"error": "invalid_balance"}), 400
set_balance(uid, balance)
return jsonify({"ok": True, "user_id": uid, "balance_abo": balance})
@APP.get("/tower/pricing")
def tower_pricing_get():
g = _tower_guard()
if g:
return g
return jsonify(get_pricing())
@APP.post("/tower/pricing/set")
def tower_pricing_set():
g = _tower_guard()
if g:
return g
data = request.get_json(silent=True) or {}
default_cost = data.get("default_op_cost_abo", None)
operations = data.get("operations", None)
cur = get_pricing()
if default_cost is not None:
try:
cur["default_op_cost_abo"] = float(default_cost)
except Exception:
return jsonify({"error": "invalid_default_cost"}), 400
if operations is not None:
if not isinstance(operations, dict):
return jsonify({"error": "invalid_operations"}), 400
cleaned = {}
for k, v in operations.items():
try:
cleaned[str(k).strip().lower()] = float(v)
except Exception:
continue
if cleaned:
cur["operations"] = cleaned
_json_save(PRICING_FILE, cur)
return jsonify({"ok": True, "pricing": cur})
# === CLIENT AUTH START ===
def find_user_by_email(email):
ensure_users_file()
email = str(email or "").strip().lower()
data = _json_load(USERS_FILE, {"users": []})
for u in data.get("users", []):
if str(u.get("email", "")).strip().lower() == email:
return u
return None
def _user_public(u):
return {
"id": str(u.get("id", "")),
"email": u.get("email", ""),
"name": u.get("name", ""),
"wallet": u.get("wallet", ""),
"plan": u.get("plan", "free"),
"ai_limit_monthly": int(u.get("ai_limit_monthly", 0) or 0),
"updated_at": u.get("updated_at", ""),
"created_at": u.get("created_at", ""),
"must_change_password": bool(u.get("must_change_password", False)),
"profile": u.get("profile", {}) if isinstance(u.get("profile", {}), dict) else {},
"postal_profile": (u.get("profile", {}) or {}).get("postal", {}) if isinstance(u.get("profile", {}), dict) else {}
}
def _pw_hash(raw):
return hashlib.sha256(str(raw or "").encode("utf-8")).hexdigest()
def _make_user_id():
return "USR-" + secrets.token_hex(4).upper()
def _client_cert_title(row: dict) -> str:
title = str(row.get("original_filename", "") or "").strip()
low = title.lower()
for ext in (".png", ".jpg", ".jpeg", ".pdf", ".txt", ".doc", ".docx"):
if low.endswith(ext):
return title[:-len(ext)]
return title
def _build_client_certificate_row(row: dict):
cert_type = str(row.get("cert_type", "base") or "base").strip()
if cert_type == "blockchain":
row = _ots_refresh_row(row)
certificate_path = str(row.get("certificate_path", "") or "").strip()
cert_id = str(row.get("cert_id", "") or "").strip()
verify_url = str(row.get("verify_url", "") or "").strip()
blockchain_status = str(row.get("blockchain_status", "") or "").strip()
if cert_type == "blockchain" and not blockchain_status:
blockchain_status = "pending"
return {
"cert_id": cert_id,
"request_id": str(row.get("request_id", "") or "").strip(),
"cert_type": cert_type,
"title": _client_cert_title(row),
"original_filename": str(row.get("original_filename", "") or "").strip(),
"author": str(row.get("author", "") or "").strip(),
"deposit_date": str(row.get("deposit_date", "") or "").strip(),
"file_hash": str(row.get("file_hash", "") or "").strip(),
"timestamp_status": str(row.get("timestamp_status", "") or "").strip(),
"timestamp_provider": str(row.get("timestamp_provider", "") or "").strip(),
"timestamp_environment": str(row.get("timestamp_environment", "") or "").strip(),
"timestamp_type": str(row.get("timestamp_type", "") or "").strip(),
"timestamp_hash": str(row.get("timestamp_hash", "") or "").strip(),
"timestamp_issued_at": str(row.get("timestamp_issued_at", "") or "").strip(),
"timestamp_transaction": str(row.get("timestamp_transaction", "") or "").strip(),
"timestamp_body_url": str(row.get("timestamp_body_url", "") or "").strip(),
"timestamp_available": row.get("timestamp_available"),
"timestamp_used": row.get("timestamp_used"),
"video_proof": bool(row.get("video_proof", False)),
"video_sha256": str(row.get("video_sha256", "") or "").strip(),
"video_declaration_text": str(row.get("video_declaration_text", "") or "").strip(),
"video_file_name": str(row.get("video_file_name", "") or "").strip(),
"video_retained_by_platform": bool(row.get("video_retained_by_platform", False)),
"verify_url": verify_url,
"certificate_path": certificate_path,
"certificate_url": _build_public_cert_url(certificate_path) if certificate_path else "",
"blockchain_status": blockchain_status,
"btc_txid": str(row.get("btc_txid", "") or "").strip(),
"btc_block_height": row.get("btc_block_height"),
"ots_path": str(row.get("ots_path", "") or "").strip(),
"ots_manifest_path": str(row.get("ots_manifest_path", "") or "").strip(),
"ots_proof_url": _build_public_cert_url(str(row.get("ots_path", "") or "").strip()) if str(row.get("ots_path", "") or "").strip() else "",
"ots_manifest_url": _build_public_cert_url(str(row.get("ots_manifest_path", "") or "").strip()) if str(row.get("ots_manifest_path", "") or "").strip() else "",
"anchored_at": str(row.get("anchored_at", "") or "").strip(),
"confirmed_at": str(row.get("confirmed_at", "") or "").strip(),
"legal_attestation": f"Certificato emesso via CopyrightChain AI Gateway. ID: {cert_id}. Verifica: {verify_url}"
}
@APP.get("/client/certificates")
def client_certificates():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
uid = str(user.get("id", "") or "").strip()
items = []
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
except Exception:
continue
if str(row.get("user_id", "") or "").strip() != uid:
continue
items.append(_build_client_certificate_row(row))
except Exception as e:
return jsonify({"error": "client_certificates_failed", "details": str(e)}), 500
items.sort(key=lambda x: str(x.get("deposit_date", "") or ""), reverse=True)
return jsonify({"ok": True, "certificates": items})
@APP.get("/tower/certificates")
def tower_certificates():
tok = request.headers.get("X-CCP-Token", "") or str(request.args.get("token", "") or "").strip()
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
q = str(request.args.get("q", "") or "").strip().lower()
status_filter = str(request.args.get("status", "") or "").strip().lower()
type_filter = str(request.args.get("type", "") or "").strip().lower()
users_by_id = {}
try:
ensure_users_file()
users_data = _json_load(USERS_FILE, {"users": []})
for u in users_data.get("users", []):
uid = str(u.get("id", "") or "").strip()
if uid:
users_by_id[uid] = u
except Exception:
users_by_id = {}
items = []
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
except Exception:
continue
cert = _build_client_certificate_row(row)
uid = str(row.get("user_id", "") or "").strip()
user = users_by_id.get(uid, {})
cert["user_id"] = uid
cert["user_email"] = str(user.get("email", "") or "").strip()
cert["user_name"] = str(user.get("name", "") or "").strip()
searchable = " ".join([
str(cert.get("cert_id", "") or ""),
str(cert.get("request_id", "") or ""),
str(cert.get("title", "") or ""),
str(cert.get("original_filename", "") or ""),
str(cert.get("file_hash", "") or ""),
str(cert.get("btc_txid", "") or ""),
str(cert.get("user_id", "") or ""),
str(cert.get("user_email", "") or ""),
str(cert.get("user_name", "") or "")
]).lower()
if q and q not in searchable:
continue
if status_filter and str(cert.get("blockchain_status", "") or "").strip().lower() != status_filter:
continue
if type_filter and str(cert.get("cert_type", "") or "").strip().lower() != type_filter:
continue
items.append(cert)
except Exception as e:
return jsonify({"error": "tower_certificates_failed", "details": str(e)}), 500
items.sort(key=lambda x: str(x.get("deposit_date", "") or ""), reverse=True)
return jsonify({"ok": True, "count": len(items), "certificates": items})
@APP.post("/client/register")
def client_register():
data = request.get_json(silent=True) or {}
name = str(data.get("name", "")).strip()
email = str(data.get("email", "")).strip().lower()
password = str(data.get("password", "")).strip()
wallet = str(data.get("wallet", "")).strip()
if not name or not email or not password:
return jsonify({"error": "missing_fields"}), 400
if "@" not in email:
return jsonify({"error": "invalid_email"}), 400
if len(password) < 8:
return jsonify({"error": "weak_password"}), 400
if find_user_by_email(email):
return jsonify({"error": "email_exists"}), 409
uid = _make_user_id()
token = secrets.token_hex(24)
now = _now_iso()
user = {
"id": uid,
"token": token,
"email": email,
"name": name,
"wallet": wallet,
"plan": "free",
"ai_limit_monthly": 0,
"password_hash": _pw_hash(password),
"created_at": now,
"updated_at": now
}
upsert_user(user)
set_balance(uid, get_balance(uid))
return jsonify({
"ok": True,
"token": token,
"user": _user_public(user),
"balance_abo": get_balance(uid),
"agency": agency_plan_public(uid),
"pricing": get_pricing()
})
@APP.post("/client/login")
def client_login():
data = request.get_json(silent=True) or {}
email = str(data.get("email", "")).strip().lower()
password = str(data.get("password", "")).strip()
if not email or not password:
return jsonify({"error": "missing_fields"}), 400
user = find_user_by_email(email)
if not user:
return jsonify({"error": "unauthorized"}), 401
if str(user.get("password_hash", "")) != _pw_hash(password):
return jsonify({"error": "unauthorized"}), 401
uid = str(user.get("id", ""))
return jsonify({
"ok": True,
"token": user.get("token", ""),
"user": _user_public(user),
"must_change_password": bool(user.get("must_change_password", False)),
"balance_abo": get_balance(uid),
"agency": agency_plan_public(uid),
"pricing": get_pricing()
})
@APP.post("/client/logout")
def client_logout():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
user["token"] = secrets.token_hex(24)
user["updated_at"] = _now_iso()
upsert_user(user)
return jsonify({"ok": True})
@APP.post("/client/ai/advisor")
def client_ai_advisor():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
prompt = str(data.get("prompt", "") or "").strip()
if not prompt:
return jsonify({"error": "missing_prompt"}), 400
if not OPENAI_API_KEY:
return jsonify({"error": "openai_not_configured"}), 503
uid = str(user.get("id"))
op = "legal_advisor"
ok_debit, bal_before, cost_abo = debit_or_reject(uid, op)
if not ok_debit:
return jsonify({
"error": "insufficient_balance",
"op": op,
"cost_abo": cost_abo,
"balance_abo": bal_before
}), 402
system_instruction = (
"Agisci come un consulente legale preliminare specializzato in proprietà intellettuale, "
"copyright, tutela delle opere digitali e prova documentale. "
"Non presentarti come avvocato del cliente e non fingere assistenza forense definitiva. "
"Fornisci una prima analisi pratica, strutturata, prudente e professionale in lingua italiana."
)
t0 = time.time()
r = requests.post(
"https://api.openai.com/v1/responses",
headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
json={
"model": os.environ.get("OPENAI_TEXT_MODEL", "gpt-4.1-mini"),
"input": [
{
"role": "system",
"content": [{"type": "input_text", "text": system_instruction}]
},
{
"role": "user",
"content": [{"type": "input_text", "text": prompt}]
}
]
},
timeout=120
)
ms = int((time.time() - t0) * 1000)
if r.status_code >= 400:
add_balance(uid, cost_abo)
log_ai_usage(
user_id=uid,
provider="openai",
ok=False,
status=r.status_code,
ms=ms,
cost_eur=0.0,
op=op,
abo_cost=cost_abo,
error="openai_error"
)
return jsonify({"error": "openai_error", "details": r.text}), r.status_code
log_ai_usage(
user_id=uid,
provider="openai",
ok=True,
status=200,
ms=ms,
cost_eur=COST_OPENAI_EUR,
op=op,
abo_cost=cost_abo
)
try:
j = r.json()
except Exception:
j = {}
text = (j.get("output_text") or "").strip()
if not text:
out = j.get("output") or []
if isinstance(out, list):
parts = []
for it in out:
if isinstance(it, dict):
for c in (it.get("content") or []):
if isinstance(c, dict) and (c.get("text") or ""):
parts.append(c.get("text"))
text = ("".join(parts)).strip()
return jsonify({"ok": True, "text": text})
@APP.post("/client/ai/defense")
def client_ai_defense():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
prompt = str(data.get("prompt", "") or "").strip()
if not prompt:
return jsonify({"error": "missing_prompt"}), 400
if not OPENAI_API_KEY:
return jsonify({"error": "openai_not_configured"}), 503
uid = str(user.get("id"))
op = "legal_defense"
ok_debit, bal_before, cost_abo = debit_or_reject(uid, op)
if not ok_debit:
return jsonify({
"error": "insufficient_balance",
"op": op,
"cost_abo": cost_abo,
"balance_abo": bal_before
}), 402
system_instruction = (
"Agisci come un redattore legale professionale specializzato in proprietà intellettuale, "
"copyright e tutela di opere digitali. "
"Genera bozze formali, strutturate, credibili e professionali in lingua italiana. "
"Non dichiarare di essere un avvocato incaricato e non inserire fatti non forniti."
)
t0 = time.time()
r = requests.post(
"https://api.openai.com/v1/responses",
headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
json={
"model": os.environ.get("OPENAI_TEXT_MODEL", "gpt-4.1-mini"),
"input": [
{
"role": "system",
"content": [{"type": "input_text", "text": system_instruction}]
},
{
"role": "user",
"content": [{"type": "input_text", "text": prompt}]
}
]
},
timeout=120
)
ms = int((time.time() - t0) * 1000)
if r.status_code >= 400:
add_balance(uid, cost_abo)
log_ai_usage(
user_id=uid,
provider="openai",
ok=False,
status=r.status_code,
ms=ms,
cost_eur=0.0,
op=op,
abo_cost=cost_abo,
error="openai_error"
)
return jsonify({"error": "openai_error", "details": r.text}), r.status_code
log_ai_usage(
user_id=uid,
provider="openai",
ok=True,
status=200,
ms=ms,
cost_eur=COST_OPENAI_EUR,
op=op,
abo_cost=cost_abo
)
try:
j = r.json()
except Exception:
j = {}
text = (j.get("output_text") or "").strip()
if not text:
out = j.get("output") or []
if isinstance(out, list):
parts = []
for it in out:
if isinstance(it, dict):
for c in (it.get("content") or []):
if isinstance(c, dict) and (c.get("text") or ""):
parts.append(c.get("text"))
text = ("".join(parts)).strip()
return jsonify({"ok": True, "text": text})
@APP.post("/client/ai/license")
def client_ai_license():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
prompt = str(data.get("prompt", "") or "").strip()
if not prompt:
return jsonify({"error": "missing_prompt"}), 400
if not OPENAI_API_KEY:
return jsonify({"error": "openai_not_configured"}), 503
uid = str(user.get("id"))
op = "generic"
ok_debit, bal_before, cost_abo = debit_or_reject(uid, op)
if not ok_debit:
return jsonify({
"error": "insufficient_balance",
"op": op,
"cost_abo": cost_abo,
"balance_abo": bal_before
}), 402
system_instruction = (
"Agisci come un redattore contrattuale professionale specializzato in licenze d'uso, "
"proprietà intellettuale e tutela di opere digitali. "
"Genera testi formali, chiari, credibili e professionali in lingua italiana. "
"Non inventare dati fattuali non forniti e non dichiarare efficacia legale assoluta."
)
t0 = time.time()
r = requests.post(
"https://api.openai.com/v1/responses",
headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
json={
"model": os.environ.get("OPENAI_TEXT_MODEL", "gpt-4.1-mini"),
"input": [
{
"role": "system",
"content": [{"type": "input_text", "text": system_instruction}]
},
{
"role": "user",
"content": [{"type": "input_text", "text": prompt}]
}
]
},
timeout=120
)
ms = int((time.time() - t0) * 1000)
if r.status_code >= 400:
add_balance(uid, cost_abo)
log_ai_usage(
user_id=uid,
provider="openai",
ok=False,
status=r.status_code,
ms=ms,
cost_eur=0.0,
op=op,
abo_cost=cost_abo,
error="openai_error"
)
return jsonify({"error": "openai_error", "details": r.text}), r.status_code
log_ai_usage(
user_id=uid,
provider="openai",
ok=True,
status=200,
ms=ms,
cost_eur=COST_OPENAI_EUR,
op=op,
abo_cost=cost_abo
)
try:
j = r.json()
except Exception:
j = {}
text = (j.get("output_text") or "").strip()
if not text:
out = j.get("output") or []
if isinstance(out, list):
parts = []
for it in out:
if isinstance(it, dict):
for c in (it.get("content") or []):
if isinstance(c, dict) and (c.get("text") or ""):
parts.append(c.get("text"))
text = ("".join(parts)).strip()
return jsonify({"ok": True, "text": text})
@APP.post("/client/change_password")
def client_change_password():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
current_password = str(data.get("current_password", "")).strip()
new_password = str(data.get("new_password", "")).strip()
force_change = bool(user.get("must_change_password", False))
if not new_password:
return jsonify({"error": "missing_fields"}), 400
if len(new_password) < 8:
return jsonify({"error": "weak_password"}), 400
if not force_change:
if not current_password:
return jsonify({"error": "missing_fields"}), 400
if str(user.get("password_hash", "")) != _pw_hash(current_password):
return jsonify({"error": "wrong_current_password"}), 401
user["password_hash"] = _pw_hash(new_password)
user["must_change_password"] = False
user["updated_at"] = _now_iso()
upsert_user(user)
return jsonify({
"ok": True,
"user": _user_public(user)
})
@APP.get("/client/me")
def client_me():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
uid = str(user.get("id", ""))
return jsonify({
"ok": True,
"user": _user_public(user),
"balance_abo": get_balance(uid),
"agency": agency_plan_public(uid),
"pricing": get_pricing()
})
# --- CCP CLIENT PROFILE START ---
def _clean_postal_profile(raw):
raw = raw if isinstance(raw, dict) else {}
def txt(key, max_len=200):
return str(raw.get(key, "") or "").strip()[:max_len]
provincia = txt("provincia", 2).upper()
nazione = txt("nazione", 2).upper() or "IT"
cap = "".join(ch for ch in txt("cap", 5) if ch.isdigit())[:5]
return {
"ragione_sociale": txt("ragione_sociale", 200),
"nome": txt("nome", 100),
"cognome": txt("cognome", 100),
"email": txt("email", 200),
"telefono": txt("telefono", 50),
"codice_fiscale": txt("codice_fiscale", 32).upper(),
"partita_iva": txt("partita_iva", 32),
"dug": txt("dug", 30) or "via",
"indirizzo": txt("indirizzo", 200),
"civico": txt("civico", 30),
"comune": txt("comune", 120),
"cap": cap,
"provincia": provincia,
"nazione": nazione
}
@APP.get("/client/profile")
def client_profile_get():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
profile = user.get("profile", {}) if isinstance(user.get("profile", {}), dict) else {}
postal = profile.get("postal", {}) if isinstance(profile.get("postal", {}), dict) else {}
return jsonify({
"ok": True,
"profile": profile,
"postal": postal,
"user": _user_public(user)
})
@APP.post("/client/profile")
def client_profile_save():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
postal_raw = data.get("postal") or data.get("postal_profile") or data
postal = _clean_postal_profile(postal_raw)
profile = user.get("profile", {}) if isinstance(user.get("profile", {}), dict) else {}
profile["postal"] = postal
display_name = " ".join([postal.get("nome", ""), postal.get("cognome", "")]).strip()
if display_name:
user["name"] = display_name
user["profile"] = profile
user["updated_at"] = _now_iso()
upsert_user(user)
return jsonify({
"ok": True,
"profile": profile,
"postal": postal,
"user": _user_public(user)
})
# --- CCP CLIENT PROFILE END ---
# === CLIENT AUTH END ===
@APP.get("/me/balance")
def me_balance():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
uid = str(user.get("id"))
return jsonify({"ok": True, "user_id": uid, "balance_abo": get_balance(uid), "agency": agency_plan_public(uid), "pricing": get_pricing()})
@APP.get("/verify/public")
def verify_public():
code = (request.args.get("code", "") or "").strip()
cert_id = (request.args.get("id", "") or "").strip()
file_hash = (request.args.get("hash", "") or "").strip()
txid = (request.args.get("txid", "") or "").strip()
q = code or cert_id or file_hash or txid
if not q:
return jsonify({"ok": False, "error": "missing_query"}), 400
found = None
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
except Exception:
continue
row_cert_id = str(row.get("cert_id", "") or "").strip()
row_hash = str(row.get("file_hash", "") or "").strip()
row_req = str(row.get("request_id", "") or "").strip()
row_txid = str(row.get("btc_txid", "") or "").strip()
if cert_id and file_hash and not (cert_id == row_cert_id and file_hash.lower() == row_hash.lower()):
continue
if q and q in (row_cert_id, row_hash, row_req, row_txid):
found = row
break
if cert_id and cert_id == row_cert_id:
found = row
break
if file_hash and file_hash == row_hash:
found = row
break
if txid and txid == row_txid:
found = row
break
except Exception as e:
return jsonify({"ok": False, "error": "verify_scan_failed", "details": str(e)}), 500
if not found:
return jsonify({"ok": False, "error": "not_found"}), 404
certificate_path = str(found.get("certificate_path", "") or "").strip()
cert_type = str(found.get("cert_type", "base") or "base").strip()
if cert_type == "blockchain":
found = _ots_refresh_row(found)
blockchain_status = str(found.get("blockchain_status", "") or "").strip()
if cert_type == "blockchain" and not blockchain_status:
blockchain_status = "pending"
title = str(found.get("original_filename", "") or "").strip()
if title.lower().endswith(".png") or title.lower().endswith(".jpg") or title.lower().endswith(".jpeg") or title.lower().endswith(".pdf"):
title = title.rsplit(".", 1)[0]
return jsonify({
"ok": True,
"verified": True,
"query": q,
"cert_id": str(found.get("cert_id", "") or "").strip(),
"file_hash": str(found.get("file_hash", "") or "").strip(),
"timestamp_status": str(found.get("timestamp_status", "") or "").strip(),
"timestamp_provider": str(found.get("timestamp_provider", "") or "").strip(),
"timestamp_environment": str(found.get("timestamp_environment", "") or "").strip(),
"timestamp_type": str(found.get("timestamp_type", "") or "").strip(),
"timestamp_hash": str(found.get("timestamp_hash", "") or "").strip(),
"timestamp_issued_at": str(found.get("timestamp_issued_at", "") or "").strip(),
"timestamp_transaction": str(found.get("timestamp_transaction", "") or "").strip(),
"timestamp_body_url": str(found.get("timestamp_body_url", "") or "").strip(),
"timestamp_available": found.get("timestamp_available"),
"timestamp_used": found.get("timestamp_used"),
"video_proof": bool(found.get("video_proof", False)),
"video_sha256": str(found.get("video_sha256", "") or "").strip(),
"video_declaration_text": str(found.get("video_declaration_text", "") or "").strip(),
"video_file_name": str(found.get("video_file_name", "") or "").strip(),
"video_retained_by_platform": bool(found.get("video_retained_by_platform", False)),
"content_type": str(found.get("content_type", "") or "").strip(),
"ai_assisted": bool(found.get("ai_assisted", False) or str(found.get("content_type", "") or "").strip() == "ai_assisted"),
"ai_tool_used": str(found.get("ai_tool_used", "") or "").strip(),
"ai_prompt": str(found.get("ai_prompt", "") or "").strip(),
"human_contribution": str(found.get("human_contribution", "") or "").strip(),
"ai_process_notes": str(found.get("ai_process_notes", "") or "").strip(),
"ai_declaration_accepted": bool(found.get("ai_declaration_accepted", False)),
"request_id": str(found.get("request_id", "") or "").strip(),
"user_id": str(found.get("user_id", "") or "").strip(),
"title": title,
"author": str(found.get("author", "") or "").strip(),
"deposit_date": str(found.get("deposit_date", "") or "").strip(),
"cert_type": cert_type,
"blockchain_status": blockchain_status,
"btc_txid": str(found.get("btc_txid", "") or "").strip(),
"btc_block_height": found.get("btc_block_height"),
"ots_path": str(found.get("ots_path", "") or "").strip(),
"ots_manifest_path": str(found.get("ots_manifest_path", "") or "").strip(),
"ots_last_upgrade_at": str(found.get("ots_last_upgrade_at", "") or "").strip(),
"anchored_at": str(found.get("anchored_at", "") or "").strip(),
"confirmed_at": str(found.get("confirmed_at", "") or "").strip(),
"certificate_path": certificate_path,
"certificate_url": _build_public_cert_url(certificate_path) if certificate_path else "",
"ots_proof_url": _build_public_cert_url(str(found.get("ots_path", "") or "").strip()) if str(found.get("ots_path", "") or "").strip() else "",
"ots_manifest_url": _build_public_cert_url(str(found.get("ots_manifest_path", "") or "").strip()) if str(found.get("ots_manifest_path", "") or "").strip() else "",
"verify_url": str(found.get("verify_url", "") or "").strip()
})
@APP.get("/cert/file")
def cert_file():
"""
Download file certificato/proof/manifest dal server path.
Accesso consentito a:
- tower session
- admin token
- client autenticato proprietario del file richiesto
"""
path = (request.args.get("path", "") or "").strip()
if not path:
return jsonify({"error": "missing_path"}), 400
real = os.path.realpath(path)
allowed_roots = [
os.path.realpath(CERT_BASE_DIR),
os.path.realpath(CERT_BC_DIR),
os.path.realpath(CERT_TS_DIR),
os.path.realpath(OTS_WORK_DIR),
]
if not any(real == root or real.startswith(root + os.sep) for root in allowed_roots):
return jsonify({"error": "forbidden_path"}), 403
if not os.path.exists(real) or not os.path.isfile(real):
return jsonify({"error": "file_not_found"}), 404
allowed = False
tok = request.headers.get("X-CCP-Token", "")
if tower_session():
allowed = True
elif ADMIN_TOKEN and tok == ADMIN_TOKEN:
allowed = True
else:
user = find_user_by_token(tok)
if user:
uid = str(user.get("id", "") or "").strip()
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
except Exception:
continue
if str(row.get("user_id", "") or "").strip() != uid:
continue
row_paths = [
str(row.get("certificate_path", "") or "").strip(),
str(row.get("ots_path", "") or "").strip(),
str(row.get("ots_manifest_path", "") or "").strip(),
]
row_paths = [os.path.realpath(x) for x in row_paths if x]
if real in row_paths:
allowed = True
break
except Exception as e:
return jsonify({"error": "auth_scan_failed", "details": str(e)}), 500
if not allowed:
return jsonify({"error": "unauthorized"}), 401
return send_file(real, as_attachment=True, download_name=os.path.basename(real))
@APP.post("/certify")
def certify():
payload = request.get_json(silent=True) or {}
return _cert_debit_core(payload)
@APP.post("/cert/debit")
def cert_debit():
payload = request.get_json(silent=True) or {}
return _cert_debit_core(payload)
@APP.post("/billing/stripe/checkout")
def billing_stripe_checkout():
if not STRIPE_SECRET_KEY:
return jsonify({"error": "stripe_not_configured"}), 500
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
try:
abo_amount = int(data.get("abo_amount", 0) or 0)
except Exception:
abo_amount = 0
if abo_amount <= 0:
return jsonify({"error": "invalid_abo_amount"}), 400
unit_amount_cents = 490
uid = str(user.get("id", "")).strip()
email = str(user.get("email", "")).strip()
try:
session = stripe.checkout.Session.create(
mode="payment",
success_url=STRIPE_SUCCESS_URL or "https://app.copyrightchain.it/?billing=success",
cancel_url=STRIPE_CANCEL_URL or "https://app.copyrightchain.it/?billing=cancel",
customer_email=email or None,
payment_method_types=["card"],
allow_promotion_codes=True,
line_items=[{
"price_data": {
"currency": "eur",
"product_data": {
"name": f"ABO Coin Recharge ({abo_amount} ABO)",
"description": f"Wallet recharge for user {uid}"
},
"unit_amount": unit_amount_cents
},
"quantity": abo_amount
}],
metadata={
"type": "abo_recharge",
"user_id": uid,
"abo_amount": str(abo_amount)
}
)
return jsonify({"ok": True, "checkout_url": session.url, "session_id": session.id})
except Exception as e:
return jsonify({"error": "stripe_checkout_error", "detail": str(e)}), 500
@APP.post("/billing/stripe/webhook")
def billing_stripe_webhook():
if not STRIPE_SECRET_KEY or not STRIPE_WEBHOOK_SECRET:
return jsonify({"error": "stripe_not_configured"}), 500
payload = request.get_data()
sig_header = request.headers.get("Stripe-Signature", "")
try:
event = stripe.Webhook.construct_event(
payload=payload,
sig_header=sig_header,
secret=STRIPE_WEBHOOK_SECRET
)
except Exception as e:
return jsonify({"error": "invalid_webhook", "detail": str(e)}), 400
payload_json = {}
try:
payload_json = json.loads(payload.decode("utf-8"))
except Exception:
payload_json = {}
event_type = str(payload_json.get("type", "")).strip()
obj = (((payload_json.get("data") or {}).get("object")) or {})
meta = obj.get("metadata", {}) or {}
if event_type == "checkout.session.completed":
uid = str(meta.get("user_id", "")).strip()
payment_id = str(obj.get("id", "")).strip()
try:
abo_amount = int(meta.get("abo_amount", "0") or "0")
except Exception:
abo_amount = 0
print("[STRIPE WEBHOOK DEBUG] event_type=", event_type, "uid=", uid, "payment_id=", payment_id, "abo_amount=", abo_amount, flush=True)
if uid and abo_amount > 0 and payment_id:
already_done = False
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
try:
row = json.loads(line)
if row.get("kind") == "stripe_recharge" and row.get("payment_id") == payment_id:
already_done = True
break
except Exception:
continue
except Exception:
pass
if not already_done:
promo_codes, full_session = stripe_session_promo_codes(payment_id, obj)
session_email = stripe_session_email(full_session, obj)
bonus_abo = 0
bonus_code = ""
if WELCOME_PROMO_CODE in promo_codes:
if not promo_bonus_already_used(WELCOME_PROMO_CODE, uid, session_email):
bonus_abo = WELCOME_PROMO_BONUS_ABO
bonus_code = WELCOME_PROMO_CODE
else:
print("[STRIPE PROMO DEBUG] WELCOM1 already used uid=", uid, "email=", session_email, flush=True)
total_credit = abo_amount + bonus_abo
set_balance(uid, get_balance(uid) + total_credit)
cert_log_request(
kind="stripe_recharge",
payment_id=payment_id,
user_id=uid,
abo_amount=abo_amount,
promo_codes=promo_codes,
bonus_code=bonus_code,
bonus_abo=bonus_abo,
credited_abo=total_credit,
customer_email=session_email,
balance_after=get_balance(uid)
)
if bonus_abo > 0:
mark_promo_bonus_used(WELCOME_PROMO_CODE, uid, session_email, payment_id, bonus_abo)
cert_log_request(
kind="stripe_promo_bonus",
payment_id=payment_id,
user_id=uid,
promo_code=WELCOME_PROMO_CODE,
bonus_abo=bonus_abo,
customer_email=session_email,
balance_after=get_balance(uid)
)
return jsonify({"ok": True})
@APP.post("/tower/users/reset-password")
def tower_users_reset_password_hyphen():
tok = request.headers.get("X-CCP-Token", "")
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "") or "").strip()
new_password = str(data.get("new_password", "") or "").strip()
if not uid:
return jsonify({"error": "missing_user_id"}), 400
if not new_password:
return jsonify({"error": "missing_new_password"}), 400
if len(new_password) < 8:
return jsonify({"error": "weak_password"}), 400
users_data = _json_load(USERS_FILE, {"users": []})
users = users_data.get("users", [])
found = None
for u in users:
if str(u.get("id", "") or "").strip() == uid:
found = u
break
if not found:
return jsonify({"error": "user_not_found"}), 404
found["password_hash"] = _pw_hash(new_password)
found["must_change_password"] = True
found["updated_at"] = _now_iso()
_json_save(USERS_FILE, users_data)
try:
pw_log = os.path.join(DATA_DIR, "tower_password_log.jsonl")
row = {
"ts": _now_iso(),
"user_id": uid,
"action": "reset_password",
"must_change_password": True,
"actor": "tower",
}
with open(pw_log, "a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
except Exception:
pass
return jsonify({
"ok": True,
"user_id": uid,
"must_change_password": True
})
@APP.post("/tower/users/credit")
def tower_users_credit():
tok = request.headers.get("X-CCP-Token", "")
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {}
uid = str(data.get("user_id", "") or "").strip()
amount = data.get("amount", None)
note = str(data.get("note", "") or "").strip()
if not uid:
return jsonify({"error": "missing_user_id"}), 400
try:
amount = float(amount)
except Exception:
return jsonify({"error": "invalid_amount"}), 400
if amount == 0:
return jsonify({"error": "zero_amount"}), 400
users_data = _json_load(USERS_FILE, {"users": []})
users = users_data.get("users", [])
found = None
for u in users:
if str(u.get("id", "") or "").strip() == uid:
found = u
break
if not found:
return jsonify({"error": "user_not_found"}), 404
before = float(get_balance(uid) or 0)
add_balance(uid, amount)
after = float(get_balance(uid) or 0)
try:
credit_log = os.path.join(DATA_DIR, "tower_credit_log.jsonl")
row = {
"ts": _now_iso(),
"user_id": uid,
"amount": amount,
"balance_before": before,
"balance_after": after,
"note": note,
"actor": "tower",
}
with open(credit_log, "a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
except Exception:
pass
return jsonify({
"ok": True,
"user_id": uid,
"amount": amount,
"balance_before": before,
"balance_after": after
})
# === AGENCY ADMIN START ===
def _admin_user_by_uid_or_email(user_id="", email=""):
user_id = str(user_id or "").strip()
email = str(email or "").strip().lower()
ensure_users_file()
data = _json_load(USERS_FILE, {"users": []})
for u in data.get("users", []):
if user_id and str(u.get("id", "") or "").strip() == user_id:
return u
if email and str(u.get("email", "") or "").strip().lower() == email:
return u
return None
@APP.get("/admin/agency/plans")
def admin_agency_plans():
if not require_admin_token(request):
return jsonify({"ok": False, "error": "unauthorized"}), 401
ensure_users_file()
users = _json_load(USERS_FILE, {"users": []}).get("users", [])
users_by_id = {str(u.get("id", "")): u for u in users}
data = _load_agency_plans()
plans = data.get("plans", {}) or {}
out = []
for uid, plan in plans.items():
if not isinstance(plan, dict):
continue
u = users_by_id.get(str(uid), {})
summary = agency_plan_public(uid)
out.append({
"user_id": str(uid),
"email": u.get("email", ""),
"name": u.get("name", ""),
"balance_abo": get_balance(uid),
**summary
})
out.sort(key=lambda x: (not x.get("active"), str(x.get("email") or x.get("user_id"))))
return jsonify({"ok": True, "plans": out, "count": len(out)})
@APP.post("/admin/agency/plan")
def admin_agency_plan_save():
if not require_admin_token(request):
return jsonify({"ok": False, "error": "unauthorized"}), 401
body = request.get_json(silent=True) or {}
user = _admin_user_by_uid_or_email(body.get("user_id", ""), body.get("email", ""))
if not user:
return jsonify({"ok": False, "error": "user_not_found"}), 404
uid = str(user.get("id", "")).strip()
monthly = body.get("monthly_credits", 30)
used = body.get("used_credits", None)
active = body.get("active", True)
reset_used = bool(body.get("reset_used", False))
monthly = max(0, _agency_int(monthly, 30))
data = _load_agency_plans()
plans = data.setdefault("plans", {})
current = plans.get(uid, {}) if isinstance(plans.get(uid, {}), dict) else {}
if reset_used:
used_credits = 0
elif used is None:
used_credits = max(0, _agency_int(current.get("used_credits"), 0))
else:
used_credits = max(0, _agency_int(used, 0))
used_credits = min(used_credits, monthly) if monthly >= 0 else used_credits
plan = {
**current,
"active": bool(active),
"plan_name": str(body.get("plan_name") or current.get("plan_name") or "Agenzia Creativa"),
"monthly_credits": monthly,
"used_credits": used_credits,
"renewal_date": str(body.get("renewal_date", current.get("renewal_date", "")) or ""),
"notes": str(body.get("notes", current.get("notes", "")) or ""),
"updated_at": _now_iso(),
}
plans[uid] = plan
_save_agency_plans(data)
try:
_append_jsonl(AGENCY_USAGE_LOG, {
"ts": _now_iso(),
"kind": "agency_plan_update",
"user_id": uid,
"email": user.get("email", ""),
"active": plan["active"],
"monthly_credits": monthly,
"used_credits": used_credits,
"reset_used": reset_used,
"actor": "admin"
})
except Exception:
pass
return jsonify({
"ok": True,
"user_id": uid,
"email": user.get("email", ""),
"name": user.get("name", ""),
"balance_abo": get_balance(uid),
"agency": agency_plan_public(uid)
})
@APP.post("/admin/agency/plan/reset")
def admin_agency_plan_reset():
if not require_admin_token(request):
return jsonify({"ok": False, "error": "unauthorized"}), 401
body = request.get_json(silent=True) or {}
user = _admin_user_by_uid_or_email(body.get("user_id", ""), body.get("email", ""))
if not user:
return jsonify({"ok": False, "error": "user_not_found"}), 404
uid = str(user.get("id", "")).strip()
data = _load_agency_plans()
plans = data.setdefault("plans", {})
plan = plans.get(uid, {}) if isinstance(plans.get(uid, {}), dict) else {}
plan.update({
"active": bool(plan.get("active", True)),
"plan_name": str(plan.get("plan_name", "Agenzia Creativa") or "Agenzia Creativa"),
"monthly_credits": max(0, _agency_int(plan.get("monthly_credits"), 30)),
"used_credits": 0,
"updated_at": _now_iso(),
})
plans[uid] = plan
_save_agency_plans(data)
try:
_append_jsonl(AGENCY_USAGE_LOG, {"ts": _now_iso(), "kind": "agency_monthly_reset", "user_id": uid, "actor": "admin"})
except Exception:
pass
return jsonify({"ok": True, "user_id": uid, "agency": agency_plan_public(uid)})
# === AGENCY ADMIN END ===
@APP.get("/tower/users/<user_id>/certificates")
def tower_user_certificates(user_id):
tok = request.headers.get("X-CCP-Token", "")
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
uid = str(user_id or "").strip()
if not uid:
return jsonify({"error": "missing_user_id"}), 400
users_data = _json_load(USERS_FILE, {"users": []})
users = users_data.get("users", [])
target = None
for u in users:
if str(u.get("id", "")).strip() == uid:
target = u
break
if not target:
return jsonify({"error": "user_not_found"}), 404
certs = []
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except Exception:
continue
row_uid = str(row.get("user_id", "") or "").strip()
if row_uid != uid:
continue
cert_type = str(row.get("cert_type", "base") or "base").strip()
if cert_type == "blockchain":
try:
row = _ots_refresh_row(row)
except Exception:
pass
certs.append(_build_client_certificate_row(row))
certs.sort(key=lambda r: str(r.get("deposit_date", "") or r.get("ts", "") or ""), reverse=True)
except Exception:
certs = []
return jsonify({
"ok": True,
"user_id": uid,
"count": len(certs),
"certificates": certs
})
@APP.get("/tower/certificates/file")
def tower_certificate_file():
tok = request.headers.get("X-CCP-Token", "") or str(request.args.get("token", "") or "").strip()
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
raw = str(request.args.get("path", "") or "").strip()
if not raw:
return jsonify({"error": "missing_path"}), 400
real = os.path.realpath(raw)
allowed_roots = [
os.path.realpath(CERT_BASE_DIR),
os.path.realpath(CERT_BC_DIR),
os.path.realpath(CERT_TS_DIR),
]
if not any(real.startswith(root + os.sep) or real == root for root in allowed_roots):
return jsonify({"error": "forbidden_path"}), 403
if not os.path.exists(real) or not os.path.isfile(real):
return jsonify({"error": "file_not_found"}), 404
return send_file(real, as_attachment=False, download_name=os.path.basename(real))
@APP.get("/tower/users")
def tower_users():
tok = request.headers.get("X-CCP-Token", "")
if not tower_session() and not (ADMIN_TOKEN and tok == ADMIN_TOKEN):
return jsonify({"error": "unauthorized"}), 401
data = _json_load(USERS_FILE, {"users": []})
users = data.get("users", [])
if not isinstance(users, list):
users = []
cert_counts = {}
try:
if os.path.exists(CERT_USAGE_LOG):
with open(CERT_USAGE_LOG, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except Exception:
continue
uid = str(row.get("user_id", "") or "").strip()
if not uid:
continue
cert_counts[uid] = cert_counts.get(uid, 0) + 1
except Exception:
cert_counts = {}
out = []
for u in users:
uid = str(u.get("id", "") or "").strip()
out.append({
"id": uid,
"email": str(u.get("email", "") or "").strip(),
"name": str(u.get("name", "") or "").strip(),
"token": str(u.get("token", "") or "").strip(),
"wallet": str(u.get("wallet", "") or "").strip(),
"plan": str(u.get("plan", "free") or "free").strip(),
"ai_limit_monthly": int(u.get("ai_limit_monthly", 0) or 0),
"must_change_password": bool(u.get("must_change_password", False)),
"created_at": str(u.get("created_at", "") or "").strip(),
"updated_at": str(u.get("updated_at", "") or "").strip(),
"balance_abo": get_balance(uid) if uid else 0,
"cert_count": int(cert_counts.get(uid, 0)),
})
out.sort(key=lambda x: (x.get("created_at", ""), x.get("id", "")), reverse=True)
return jsonify({"ok": True, "count": len(out), "users": out})
@APP.post("/text")
def ai_text():
payload = request.get_json(silent=True) or {}
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"error": "unauthorized"}), 401
prompt = str(payload.get("prompt", "")).strip()
if not prompt:
return jsonify({"error": "missing_prompt"}), 400
op = str(payload.get("op", "generic")).strip().lower()
uid = str(user.get("id"))
ok_debit, bal_before, cost_abo = debit_or_reject(uid, op)
if not ok_debit:
return jsonify({"error": "insufficient_balance", "op": op, "cost_abo": cost_abo, "balance_abo": bal_before}), 402
t0 = time.time()
r = requests.post(
"https://api.openai.com/v1/responses",
headers={"Authorization": f"Bearer {OPENAI_API_KEY}"},
json={"model": "gpt-4.1-mini", "input": [{"role": "user", "content": [{"type": "input_text", "text": prompt}]}]},
timeout=120
)
ms = int((time.time() - t0) * 1000)
if r.status_code >= 400:
add_balance(uid, cost_abo)
log_ai_usage(user_id=uid, provider="openai", ok=False, status=r.status_code, ms=ms, cost_eur=0.0, op=op, abo_cost=cost_abo, error="openai_error")
return jsonify({"error": "openai_error", "details": r.text}), r.status_code
log_ai_usage(user_id=uid, provider="openai", ok=True, status=200, ms=ms, cost_eur=COST_OPENAI_EUR, op=op, abo_cost=cost_abo)
try:
j = r.json()
except Exception:
j = {}
text = (j.get("output_text") or "").strip()
if not text:
out = j.get("output") or []
if isinstance(out, list):
parts = []
for it in out:
if isinstance(it, dict):
for c in (it.get("content") or []):
if isinstance(c, dict) and (c.get("text") or ""):
parts.append(c.get("text"))
text = ("".join(parts)).strip()
return jsonify({"ok": True, "text": text})
# --- CCP POSTAL PRICING START ---
import json as _postal_json
import urllib.request as _postal_request
import urllib.error as _postal_error
from pathlib import Path as _postal_Path
from flask import jsonify as _postal_jsonify
def _postal_env():
data = {}
try:
for line in _postal_Path("/opt/ccp-ai-gateway/.env").read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
data[k.strip()] = v.strip()
except Exception:
pass
postal_env = str(data.get("OPENAPI_POSTAL_ENV", "sandbox") or "sandbox").strip().lower()
if postal_env in ["production", "prod", "live"]:
prod_token = str(data.get("OPENAPI_POSTAL_PROD_TOKEN", "") or "").strip()
prod_base = str(data.get("OPENAPI_POSTAL_PROD_BASE_URL", "") or "").strip()
if prod_token:
data["OPENAPI_POSTAL_TOKEN"] = prod_token
if prod_base:
data["OPENAPI_POSTAL_BASE_URL"] = prod_base.rstrip("/")
return data
@APP.route("/api/ai/postal/pricing", methods=["GET"])
def ccp_postal_pricing():
env = _postal_env()
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
if not token:
return _postal_jsonify({"success": False, "error": "OPENAPI_POSTAL_TOKEN missing"}), 500
req = _postal_request.Request(
base + "/pricing/",
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"User-Agent": "curl/8.0"
},
method="GET"
)
try:
with _postal_request.urlopen(req, timeout=40) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
data = _postal_json.loads(raw)
except Exception:
return _postal_jsonify({"success": False, "error": "Risposta non JSON", "raw": raw[:1000]}), status
services = []
for item in data.get("data", []) or []:
first = (item.get("tariffe") or [{}])[0]
postale = float(first.get("tariffa_postale") or 0)
stampa = float(first.get("stampa") or 0)
imbustamento = float(first.get("imbustamento") or 0)
services.append({
"prodotto": item.get("prodotto"),
"descrizione": item.get("descrizione"),
"attivo": item.get("attivo"),
"accetta_allegati": item.get("accetta_allegati"),
"costo_1_pagina": round(postale + stampa + imbustamento, 2)
})
return _postal_jsonify({
"success": data.get("success"),
"message": data.get("message"),
"error": data.get("error"),
"services": services
}), status
# --- CCP POSTAL PRICING END ---
# --- CCP POSTAL SMART DRAFT START ---
from flask import request as _postal_draft_request_flask
import hmac as _postal_draft_hmac
import time as _postal_draft_time
from pathlib import Path as _postal_draft_Path
def _postal_admin_ok():
env = _postal_env()
expected = env.get("CCP_POSTAL_ADMIN_KEY", "")
received = _postal_draft_request_flask.headers.get("X-CCP-Admin-Key", "")
return bool(expected) and _postal_draft_hmac.compare_digest(expected, received)
@APP.route("/api/ai/postal/smart/draft", methods=["POST"])
def ccp_postal_smart_draft():
if not _postal_admin_ok():
return _postal_jsonify({"success": False, "error": "unauthorized"}), 401
env = _postal_env()
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
body = _postal_draft_request_flask.get_json(silent=True) or {}
body.setdefault("opzioni", {})
body["opzioni"]["autoconfirm"] = False
payload = _postal_json.dumps(body, ensure_ascii=False).encode("utf-8")
req = _postal_request.Request(
base + "/raccomandate_smart/",
data=payload,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "curl/8.0"
},
method="POST"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
response = _postal_json.loads(raw)
except Exception:
response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
# Prezzo pubblico CopyrightChain: il costo Openapi resta interno/admin.
try:
service_price = float(env.get("CCP_POSTAL_SMART_PUBLIC_PRICE_EUR", "9.90") or 9.90)
except Exception:
service_price = 9.90
try:
provider_cost = None
data_obj = response.get("data") if isinstance(response, dict) else None
first_obj = data_obj[0] if isinstance(data_obj, list) and data_obj else data_obj
if isinstance(first_obj, dict):
provider_cost = (((first_obj.get("pricing") or {}).get("totale") or {}).get("importo_totale"))
except Exception:
provider_cost = None
try:
d = _postal_draft_Path(data_dir)
d.mkdir(parents=True, exist_ok=True)
log = {
"ts": int(_postal_draft_time.time()),
"kind": "raccomandata_smart_draft",
"status": status,
"success": response.get("success"),
"response": response
}
with (d / "postal_smart_drafts.jsonl").open("a", encoding="utf-8") as f:
f.write(_postal_json.dumps(log, ensure_ascii=False) + "\n")
except Exception:
pass
return _postal_jsonify(response), status
# --- CCP POSTAL SMART DRAFT END ---
# --- CCP POSTAL SMART CONFIRM START ---
@APP.route("/api/ai/postal/smart/confirm/<postal_id>", methods=["PATCH", "POST"])
def ccp_postal_smart_confirm(postal_id):
if not _postal_admin_ok():
return _postal_jsonify({"success": False, "error": "unauthorized"}), 401
env = _postal_env()
postal_env = env.get("OPENAPI_POSTAL_ENV", "sandbox").lower()
if postal_env != "sandbox":
return _postal_jsonify({
"success": False,
"error": "production_blocked",
"message": "Conferma bloccata: questo endpoint è abilitato solo in sandbox."
}), 403
body = _postal_draft_request_flask.get_json(silent=True) or {}
if body.get("confirm_phrase") != "CONFIRM_SANDBOX_POSTAL_SEND":
return _postal_jsonify({
"success": False,
"error": "missing_confirm_phrase",
"required": "CONFIRM_SANDBOX_POSTAL_SEND"
}), 400
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
payload = _postal_json.dumps({"confirmed": True}, ensure_ascii=False).encode("utf-8")
req = _postal_request.Request(
base + "/raccomandate_smart/" + postal_id,
data=payload,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "curl/8.0"
},
method="PATCH"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
response = _postal_json.loads(raw)
except Exception:
response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
return _postal_jsonify(response), status
# --- CCP POSTAL SMART CONFIRM END ---
# --- CCP POSTAL SMART DETAIL START ---
@APP.route("/api/ai/postal/smart/detail/<postal_id>", methods=["GET"])
def ccp_postal_smart_detail(postal_id):
if not _postal_admin_ok():
return _postal_jsonify({"success": False, "error": "unauthorized"}), 401
env = _postal_env()
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
req = _postal_request.Request(
base + "/raccomandate_smart/" + postal_id,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"User-Agent": "curl/8.0"
},
method="GET"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
response = _postal_json.loads(raw)
except Exception:
response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
return _postal_jsonify(response), status
# --- CCP POSTAL SMART DETAIL END ---
# --- CCP ADMIN POSTAL SMART DRAFTS START ---
@APP.route("/admin/postal/smart/drafts", methods=["GET"])
def admin_postal_smart_drafts():
admin_payload = require_admin_token(request)
if not admin_payload:
return jsonify({"ok": False, "error": "unauthorized"}), 401
try:
env = _postal_env()
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
d = _postal_draft_Path(data_dir)
f = d / "postal_client_smart_drafts.jsonl"
items = []
if f.exists():
for line in f.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
row = _postal_json.loads(line)
response = row.get("response") if isinstance(row.get("response"), dict) else {}
data = response.get("data") if isinstance(response, dict) else {}
if isinstance(data, list):
data = data[0] if data else {}
item = {
"ts": row.get("ts"),
"user_id": row.get("user_id"),
"user_email": row.get("user_email"),
"postal_id": row.get("postal_id") or (data.get("id") if isinstance(data, dict) else ""),
"state": row.get("state") or (data.get("state") if isinstance(data, dict) else ""),
"confirmed": row.get("confirmed"),
"canceled": row.get("canceled"),
"admin_canceled": row.get("admin_canceled"),
"canceled_at": row.get("canceled_at"),
"success": row.get("success"),
"provider_cost_eur": row.get("provider_cost_eur"),
"public_price_eur": row.get("public_price_eur"),
"service_mode": row.get("service_mode"),
"created_at": row.get("ts"),
"raw": row
}
items.append(item)
except Exception:
continue
items = sorted(items, key=lambda x: int(x.get("ts") or 0), reverse=True)[:200]
return jsonify({"ok": True, "drafts": items, "count": len(items)})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
# --- CCP ADMIN POSTAL SMART DRAFTS END ---
# --- CCP ADMIN POSTAL SMART CANCEL START ---
@APP.route("/admin/postal/smart/cancel", methods=["POST"])
def admin_postal_smart_cancel():
admin_payload = require_admin_token(request)
if not admin_payload:
return jsonify({"ok": False, "error": "unauthorized"}), 401
body = request.get_json(silent=True) or {}
postal_id = str(body.get("postal_id", "") or "").strip()
ts = str(body.get("ts", "") or "").strip()
if not postal_id and not ts:
return jsonify({"ok": False, "error": "missing_postal_id_or_ts"}), 400
try:
env = _postal_env()
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
d = _postal_draft_Path(data_dir)
f = d / "postal_client_smart_drafts.jsonl"
if not f.exists():
return jsonify({"ok": False, "error": "draft_log_not_found"}), 404
lines = f.read_text(encoding="utf-8").splitlines()
found = False
canceled_row = None
out = []
for line in lines:
if not line.strip():
out.append(line)
continue
try:
row = _postal_json.loads(line)
except Exception:
out.append(line)
continue
response = row.get("response") if isinstance(row.get("response"), dict) else {}
data = response.get("data") if isinstance(response, dict) else {}
if isinstance(data, list):
data = data[0] if data else {}
row_postal_id = str(row.get("postal_id") or (data.get("id") if isinstance(data, dict) else "") or "").strip()
row_ts = str(row.get("ts", "") or "").strip()
match = False
if postal_id and row_postal_id == postal_id:
match = True
elif ts and row_ts == ts:
match = True
if match:
if bool(row.get("admin_confirmed")) or bool(row.get("confirmed")):
return jsonify({
"ok": False,
"error": "already_confirmed",
"message": "Bozza già confermata: non può essere annullata dal registro."
}), 409
row["admin_canceled"] = True
row["canceled"] = True
row["canceled_at"] = int(_postal_draft_time.time())
row["canceled_by"] = "admin"
row["state"] = "CANCELED"
found = True
canceled_row = row
out.append(_postal_json.dumps(row, ensure_ascii=False))
else:
out.append(line)
if not found:
return jsonify({"ok": False, "error": "draft_not_found"}), 404
f.write_text("\n".join(out) + "\n", encoding="utf-8")
return jsonify({
"ok": True,
"canceled": True,
"postal_id": postal_id,
"ts": ts,
"draft": canceled_row
})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
# --- CCP ADMIN POSTAL SMART CANCEL END ---
# --- CCP ADMIN POSTAL SMART CONFIRM AND DEBIT START ---
@APP.route("/admin/postal/smart/confirm", methods=["POST"])
def admin_postal_smart_confirm_and_debit():
admin_payload = require_admin_token(request)
if not admin_payload:
return jsonify({"ok": False, "error": "unauthorized"}), 401
body = request.get_json(silent=True) or {}
postal_id = str(body.get("postal_id", "")).strip()
if not postal_id:
return jsonify({"ok": False, "error": "missing_postal_id"}), 400
try:
env = _postal_env()
postal_env = env.get("OPENAPI_POSTAL_ENV", "sandbox").lower()
production_confirm_enabled = str(env.get("CCP_POSTAL_PRODUCTION_CONFIRM_ENABLED", "0")).lower() in ["1", "true", "yes", "on"]
if postal_env != "sandbox" and not production_confirm_enabled:
return jsonify({
"ok": False,
"error": "production_blocked",
"message": "Conferma invio reale bloccata. Impostare CCP_POSTAL_PRODUCTION_CONFIRM_ENABLED=1 solo dopo test e controllo credenziali produzione."
}), 403
if postal_env != "sandbox":
if str(body.get("confirm_phrase", "")).strip() != "CONFIRM_REAL_POSTAL_SEND":
return jsonify({
"ok": False,
"error": "missing_real_confirm_phrase",
"required": "CONFIRM_REAL_POSTAL_SEND"
}), 400
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
d = _postal_draft_Path(data_dir)
f = d / "postal_client_smart_drafts.jsonl"
if not f.exists():
return jsonify({"ok": False, "error": "draft_log_not_found"}), 404
lines = f.read_text(encoding="utf-8").splitlines()
rows = []
target_index = None
target = None
for idx, line in enumerate(lines):
if not line.strip():
rows.append(None)
continue
try:
row = _postal_json.loads(line)
except Exception:
rows.append(None)
continue
response = row.get("response") if isinstance(row.get("response"), dict) else {}
data = response.get("data") if isinstance(response, dict) else {}
if isinstance(data, list):
data = data[0] if data else {}
row_postal_id = str(row.get("postal_id") or (data.get("id") if isinstance(data, dict) else "") or "").strip()
rows.append(row)
if row_postal_id == postal_id:
target_index = idx
target = row
if target is None or target_index is None:
return jsonify({"ok": False, "error": "draft_not_found", "postal_id": postal_id}), 404
if bool(target.get("admin_canceled")) or bool(target.get("canceled")) or str(target.get("state", "")).upper() == "CANCELED":
return jsonify({
"ok": False,
"error": "draft_canceled",
"message": "Bozza annullata: non può essere confermata."
}), 409
if bool(target.get("admin_confirmed")):
return jsonify({
"ok": True,
"already_confirmed": True,
"postal_id": postal_id,
"abo_charged": target.get("abo_charged"),
"balance_after": target.get("balance_after")
})
user_id = str(target.get("user_id", "")).strip()
if not user_id:
return jsonify({"ok": False, "error": "missing_user_id_in_draft"}), 400
service_mode = str(target.get("service_mode") or "").strip()
cost_abo = 3.0 if service_mode == "certifica_invia" else 2.0
balance_before = get_balance(user_id)
if balance_before < cost_abo:
return jsonify({
"ok": False,
"error": "insufficient_balance",
"user_id": user_id,
"balance_abo": balance_before,
"cost_abo": cost_abo
}), 402
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
payload = _postal_json.dumps({"confirmed": True}, ensure_ascii=False).encode("utf-8")
req = _postal_request.Request(
base + "/raccomandate_smart/" + postal_id,
data=payload,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "curl/8.0"
},
method="PATCH"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
try:
confirm_response = _postal_json.loads(raw)
except Exception:
confirm_response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
if status < 200 or status >= 300 or confirm_response.get("success") is False:
return jsonify({
"ok": False,
"error": "openapi_confirm_failed",
"status": status,
"response": confirm_response
}), 502
balance_after = balance_before - cost_abo
set_balance(user_id, balance_after)
target["admin_confirmed"] = True
target["confirmed"] = True
target["confirmed_at"] = int(_postal_draft_time.time())
target["confirmed_by"] = "admin"
target["abo_charged"] = cost_abo
target["balance_before"] = balance_before
target["balance_after"] = balance_after
target["confirm_status"] = status
target["confirm_response"] = confirm_response
try:
data = confirm_response.get("data") if isinstance(confirm_response, dict) else {}
if isinstance(data, list):
data = data[0] if data else {}
if isinstance(data, dict):
target["state"] = data.get("state") or target.get("state")
except Exception:
pass
rows[target_index] = target
out_lines = []
for idx, original in enumerate(lines):
if idx == target_index:
out_lines.append(_postal_json.dumps(target, ensure_ascii=False))
else:
out_lines.append(original)
f.write_text("\n".join(out_lines) + "\n", encoding="utf-8")
return jsonify({
"ok": True,
"postal_id": postal_id,
"user_id": user_id,
"service_mode": service_mode,
"abo_charged": cost_abo,
"balance_before": balance_before,
"balance_after": balance_after,
"openapi_status": status,
"openapi_response": confirm_response
})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
# --- CCP ADMIN POSTAL SMART CONFIRM AND DEBIT END ---
# --- CCP CLIENT POSTAL SMART HISTORY START ---
@APP.route("/client/postal/smart/history", methods=["GET"])
def ccp_client_postal_smart_history():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return jsonify({"ok": False, "error": "unauthorized"}), 401
uid = str(user.get("id", "")).strip()
try:
env = _postal_env()
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
d = _postal_draft_Path(data_dir)
f = d / "postal_client_smart_drafts.jsonl"
items = []
if f.exists():
for line in f.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
row = _postal_json.loads(line)
except Exception:
continue
if str(row.get("user_id", "")).strip() != uid:
continue
response = row.get("response") if isinstance(row.get("response"), dict) else {}
data = response.get("data") if isinstance(response, dict) else {}
if isinstance(data, list):
data = data[0] if data else {}
postal_id = row.get("postal_id") or (data.get("id") if isinstance(data, dict) else "")
state = row.get("state") or (data.get("state") if isinstance(data, dict) else "")
confirmed = bool(row.get("confirmed") or row.get("admin_confirmed"))
recipient = row.get("recipient") if isinstance(row.get("recipient"), dict) else {}
cert = row.get("certificate") if isinstance(row.get("certificate"), dict) else {}
recipient_display = (
str(recipient.get("ragione_sociale") or "").strip()
or (str(recipient.get("nome") or "").strip() + " " + str(recipient.get("cognome") or "").strip()).strip()
or str(recipient.get("email") or "").strip()
or "—"
)
certificate_id = str(cert.get("id") or "").strip()
certificate_verify_url = str(cert.get("verify_url") or "").strip()
items.append({
"ts": row.get("ts"),
"created_at": row.get("ts"),
"postal_id": postal_id,
"state": state,
"confirmed": confirmed,
"service_mode": row.get("service_mode"),
"public_price_eur": row.get("public_price_eur"),
"abo_charged": row.get("abo_charged"),
"balance_after": row.get("balance_after"),
"certificate": cert if cert else None,
"certificate_id": certificate_id,
"certificate_verify_url": certificate_verify_url,
"legacy": not bool(certificate_id),
"success": row.get("success"),
"title": row.get("title") or "Invio documento",
"recipient": recipient,
"recipient_display": recipient_display,
"pdf_url": row.get("pdf_url") or row.get("document_url") or "",
})
items = sorted(items, key=lambda x: int(x.get("ts") or 0), reverse=True)[:200]
return jsonify({"ok": True, "items": items, "count": len(items)})
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 500
# --- CCP CLIENT POSTAL SMART HISTORY END ---
# --- CCP CLIENT POSTAL SMART CREATE START ---
def _ccp_validate_postal_recipient(recipient):
if not isinstance(recipient, dict):
return ["recipient"]
def v(k):
return str(recipient.get(k) or "").strip()
errors = []
has_person = bool(v("nome") and v("cognome"))
has_company = bool(v("ragione_sociale"))
if not has_person and not has_company:
errors.append("nome_cognome_o_ragione_sociale")
required = {
"dug": "tipo_strada",
"indirizzo": "indirizzo",
"civico": "civico",
"comune": "comune",
"cap": "cap",
"provincia": "provincia",
"nazione": "nazione",
}
for key, label in required.items():
if not v(key):
errors.append(label)
cap = v("cap")
if cap and (not cap.isdigit() or len(cap) != 5):
errors.append("cap_non_valido")
provincia = v("provincia").upper()
if provincia and (len(provincia) != 2 or not provincia.isalpha()):
errors.append("provincia_non_valida")
nazione = v("nazione").upper()
if nazione and len(nazione) != 2:
errors.append("nazione_non_valida")
return errors
@APP.route("/client/postal/smart/create", methods=["POST"])
def ccp_client_postal_smart_create():
tok = request.headers.get("X-CCP-Token", "")
user = find_user_by_token(tok)
if not user:
return _postal_jsonify({"success": False, "error": "unauthorized"}), 401
body = request.get_json(silent=True) or {}
profile = user.get("profile", {}) if isinstance(user.get("profile", {}), dict) else {}
saved_postal = profile.get("postal", {}) if isinstance(profile.get("postal", {}), dict) else {}
sender = body.get("sender") or saved_postal or {}
recipient = body.get("recipient") or {}
recipient_errors = _ccp_validate_postal_recipient(recipient)
if recipient_errors:
return jsonify({
"ok": False,
"success": False,
"error": "recipient_validation_failed",
"message": "Dati destinatario incompleti o non validi.",
"fields": recipient_errors
}), 400
title = body.get("title") or "Invio documento online"
text = body.get("text") or "Documento inviato tramite CopyrightChain."
mode = body.get("mode") or "solo_invio"
certificate = body.get("certificate") or {}
document_lines = []
document_lines.append(title)
document_lines.append("")
document_lines.append(text)
if mode in ["certifica_invia", "certifica_e_invia", "certified"] and certificate:
document_lines.append("")
document_lines.append("PROVA DIGITALE COPYRIGHTCHAIN")
document_lines.append("ID certificato: " + str(certificate.get("id", "")))
document_lines.append("Titolo: " + str(certificate.get("title", "")))
document_lines.append("Hash SHA-256: " + str(certificate.get("hash", "")))
document_lines.append("Verifica pubblica: " + str(certificate.get("verify_url", "")))
document_lines.append("Provider: " + str(certificate.get("provider", "")))
openapi_body = {
"mittente": sender,
"destinatari": [recipient],
"documento": ["\n".join(document_lines)],
"opzioni": {
"fronteretro": bool(body.get("fronteretro", False)),
"colori": bool(body.get("colori", False)),
"ar": bool(body.get("ar", True)),
"autoconfirm": False
}
}
env = _postal_env()
postal_env = str(env.get("OPENAPI_POSTAL_ENV", "sandbox") or "sandbox").strip().lower()
production_create_enabled = str(env.get("CCP_POSTAL_PRODUCTION_CREATE_ENABLED", "0")).lower() in ["1", "true", "yes", "on"]
if postal_env != "sandbox" and not production_create_enabled:
return _postal_jsonify({
"success": False,
"error": "production_create_blocked",
"message": "Creazione invio reale bloccata. Abilitare CCP_POSTAL_PRODUCTION_CREATE_ENABLED=1 solo per test controllato o produzione."
}), 403
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
data_dir = env.get("CCP_POSTAL_DATA_DIR", "/opt/ccp-ai-gateway/data/postal")
payload = _postal_json.dumps(openapi_body, ensure_ascii=False).encode("utf-8")
req = _postal_request.Request(
base + "/raccomandate_smart/",
data=payload,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "curl/8.0"
},
method="POST"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
response = _postal_json.loads(raw)
except Exception:
response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
# Prezzo pubblico CopyrightChain: il costo Openapi resta interno/admin.
premium_mode = mode in ["certifica_invia", "certifica_e_invia", "certified"]
price_env_key = "CCP_POSTAL_SMART_PREMIUM_PRICE_EUR" if premium_mode else "CCP_POSTAL_SMART_PUBLIC_PRICE_EUR"
default_price = "14.90" if premium_mode else "9.90"
try:
service_price = float(env.get(price_env_key, default_price) or default_price)
except Exception:
service_price = float(default_price)
try:
provider_cost = None
data_obj = response.get("data") if isinstance(response, dict) else None
first_obj = data_obj[0] if isinstance(data_obj, list) and data_obj else data_obj
if isinstance(first_obj, dict):
provider_cost = (((first_obj.get("pricing") or {}).get("totale") or {}).get("importo_totale"))
except Exception:
provider_cost = None
try:
d = _postal_draft_Path(data_dir)
d.mkdir(parents=True, exist_ok=True)
response_data_obj = response.get("data") if isinstance(response, dict) else None
response_first_obj = response_data_obj[0] if isinstance(response_data_obj, list) and response_data_obj else response_data_obj
if not isinstance(response_first_obj, dict):
response_first_obj = {}
postal_id_value = str(
response_first_obj.get("id")
or response_first_obj.get("_id")
or response_first_obj.get("uid")
or ""
).strip()
state_value = str(
response_first_obj.get("state")
or response_first_obj.get("status")
or ""
).strip()
service_mode_value = "certifica_invia" if premium_mode else "solo_invio"
log = {
"ts": int(_postal_draft_time.time()),
"kind": "client_raccomandata_smart_draft",
"user_id": str(user.get("id", "")),
"user_email": str(user.get("email", "")),
"status": status,
"success": response.get("success") if isinstance(response, dict) else None,
"postal_id": postal_id_value,
"confirmed": response_first_obj.get("confirmed"),
"admin_confirmed": False,
"state": state_value,
"title": title,
"sender": sender,
"recipient": recipient,
"certificate": certificate if isinstance(certificate, dict) and certificate else None,
"mode": mode,
"service_mode": service_mode_value,
"provider_cost_eur": provider_cost,
"public_price_eur": service_price,
"response": response
}
with (d / "postal_client_smart_drafts.jsonl").open("a", encoding="utf-8") as f:
f.write(_postal_json.dumps(log, ensure_ascii=False) + "\n")
except Exception:
pass
# Risposta pubblica al cliente: rimuove il dettaglio pricing Openapi.
public_response = response
try:
public_response = _postal_json.loads(_postal_json.dumps(response, ensure_ascii=False))
public_response["service_price_eur"] = service_price
public_response["service_mode"] = "certifica_invia" if premium_mode else "solo_invio"
data_obj = public_response.get("data")
items = data_obj if isinstance(data_obj, list) else [data_obj]
for item in items:
if isinstance(item, dict):
item.pop("pricing", None)
item["service_price_eur"] = service_price
item["service_mode"] = "certifica_invia" if premium_mode else "solo_invio"
item["confirmed"] = bool(item.get("confirmed", False))
except Exception:
public_response = response
return _postal_jsonify(public_response), status
# --- CCP CLIENT POSTAL SMART CREATE END ---
# --- CCP POSTAL SMART SIMPLE CREATE START ---
@APP.route("/api/ai/postal/smart/create", methods=["POST"])
def ccp_postal_smart_simple_create():
if not _postal_admin_ok():
return _postal_jsonify({"success": False, "error": "unauthorized"}), 401
body = _postal_draft_request_flask.get_json(silent=True) or {}
sender = body.get("sender") or {}
recipient = body.get("recipient") or {}
recipient_errors = _ccp_validate_postal_recipient(recipient)
if recipient_errors:
return jsonify({
"ok": False,
"success": False,
"error": "recipient_validation_failed",
"message": "Dati destinatario incompleti o non validi.",
"fields": recipient_errors
}), 400
title = body.get("title") or "Invio documento online"
text = body.get("text") or "Documento inviato tramite CopyrightChain."
mode = body.get("mode") or "solo_invio"
certificate = body.get("certificate") or {}
document_lines = []
document_lines.append(title)
document_lines.append("")
document_lines.append(text)
if mode in ["certifica_invia", "certifica_e_invia", "certified"] and certificate:
document_lines.append("")
document_lines.append("PROVA DIGITALE COPYRIGHTCHAIN")
document_lines.append("ID certificato: " + str(certificate.get("id", "")))
document_lines.append("Titolo: " + str(certificate.get("title", "")))
document_lines.append("Hash SHA-256: " + str(certificate.get("hash", "")))
document_lines.append("Verifica pubblica: " + str(certificate.get("verify_url", "")))
document_lines.append("Provider: " + str(certificate.get("provider", "")))
openapi_body = {
"mittente": sender,
"destinatari": [recipient],
"documento": ["\n".join(document_lines)],
"opzioni": {
"fronteretro": bool(body.get("fronteretro", False)),
"colori": bool(body.get("colori", False)),
"ar": bool(body.get("ar", True)),
"autoconfirm": False
}
}
env = _postal_env()
token = env.get("OPENAPI_POSTAL_TOKEN", "")
base = env.get("OPENAPI_POSTAL_BASE_URL", "https://test.ws.ufficiopostale.com").rstrip("/")
payload = _postal_json.dumps(openapi_body, ensure_ascii=False).encode("utf-8")
req = _postal_request.Request(
base + "/raccomandate_smart/",
data=payload,
headers={
"Authorization": "Bearer " + token,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "curl/8.0"
},
method="POST"
)
try:
with _postal_request.urlopen(req, timeout=45) as r:
raw = r.read().decode("utf-8", "replace")
status = r.getcode()
except _postal_error.HTTPError as e:
raw = e.read().decode("utf-8", "replace")
status = e.code
except Exception as e:
return _postal_jsonify({"success": False, "error": str(e)}), 500
try:
response = _postal_json.loads(raw)
except Exception:
response = {"success": False, "error": "Risposta non JSON", "raw": raw[:2000]}
return _postal_jsonify(response), status
# --- CCP POSTAL SMART SIMPLE CREATE END ---
if __name__ == "__main__":
APP.run("127.0.0.1", 5060)