import json
import re
import sqlite3
import random
from datetime import date, datetime, timedelta
from pathlib import Path
from urllib.parse import quote
from flask import Blueprint, abort, current_app, flash, g, redirect, render_template, request, session, url_for
from werkzeug.utils import secure_filename
try:
from PIL import Image, ImageOps
except ImportError:
Image = None
ImageOps = None
TODAY_ISO = date.today().isoformat()
LOCATION_VISIBILITY = ("full", "city", "country", "custom", "hidden")
LOCATION_TYPES = ("owned warehouse", "partner warehouse", "supplier location", "secure storage", "other")
INSPECTION_STATUSES = ("not checked", "inspection pending", "visually checked", "mechanically inspected", "third-party inspected", "documentation only")
SERIAL_STATUSES = ("available publicly", "available on request", "recorded internally", "pending verification", "unavailable")
CONDITION_STATUSES = ("not checked", "visually checked", "mechanically checked", "third-party checked", "documentation only", "pending")
GALLERY_CATEGORIES = ("company premises", "storefront", "warehouse", "machinery storage", "machinery loading", "transport", "inspection", "team", "delivery")
TRUST_INTERFACE_STRINGS = {
"verified_uk_company": "Verified Canadian Dealer",
"verification_intro": "{company_name} is an OMVIC-registered Canadian dealer. You can verify our dealer license information.",
"verify_companies_house": "Verify dealer license",
"company_number": "Dealer license number",
"vat_number": "HST number",
"registered_office": "Registered office",
"sales_phone": "Sales phone",
"sales_email": "Sales email",
"current_machine_location": "Current vehicle location",
"inspection_or_collection": "Inspection or collection",
"available_by_appointment": "Available by appointment",
"delivery_insured_europe": "Insured delivery available across Canada",
"listing_verification": "Listing Verification",
"listing_information_verified": "Availability and listing details last confirmed: {date}",
"serial_number_status": "Serial number",
"inspection_status": "Inspection status",
"condition_check_status": "Condition check",
"inspection_date": "Inspection date",
"availability_checked_today": "Availability checked: {date}",
"inspection_not_checked": "Not checked",
"inspection_pending": "Inspection pending",
"inspection_visually_checked": "Visually checked",
"inspection_mechanically_inspected": "Mechanically inspected",
"inspection_third_party_inspected": "Third-party inspected",
"inspection_documentation_only": "Documentation only",
"condition_not_checked": "Not checked",
"condition_visually_checked": "Visually checked",
"condition_mechanically_checked": "Mechanically checked",
"condition_third_party_checked": "Third-party checked",
"condition_documentation_only": "Documentation only",
"condition_pending": "Pending",
"collection_available": "Collection available",
"delivery_available": "Delivery available",
"yes": "Yes",
"no": "No",
"vat_short": "HST applied according to your province of registration.",
"vat_cross_border_title": "HST and licensing",
"vat_general_guidance": "HST is applied according to your province of registration. All pricing is in Canadian dollars (CAD). There are no hidden customs or import fees.",
"vat_business_guidance": "For business purchases, HST may be recoverable through your regular input tax credit filings. We provide a proper invoice with our HST number.",
"vat_private_guidance": "For private purchases, applicable HST and licensing fees will be shown clearly in the final quotation.",
"vat_final_quote": "The final quotation and invoice will confirm HST, licensing, and any dealer fees for the specific transaction.",
"vat_disclaimer": "This information is general guidance and does not replace professional tax advice.",
"company_purchase": "Company purchase",
"private_purchase": "Private purchase",
"delivery_postcode": "Delivery postal code",
"country": "Country",
"preferred_contact_method": "Preferred contact method",
"email_contact": "Email",
"telephone_contact": "Telephone",
"written_support_languages": "Telephone support is currently available in English. Written enquiries can be handled in supported website languages using our translated enquiry system.",
"team": "Team",
"recent_deliveries": "Recent Deliveries",
"purchase_process": "Purchase Process",
"company_gallery": "Company Gallery",
"machine_location_not_public": "Vehicle location available on request",
"last_checked_inventory": "Last checked inventory",
"trust_gallery": "Real Company & Warehouse Photos",
"trust_gallery_intro": "Real photographs from {COMPANY_NAME} premises, storage, loading, inspection and delivery operations.",
"buyer_type": "Buyer type",
"listing_whatsapp_message": "Hello, I am interested in {LISTING_TITLE}, reference {CAS_REFERENCE}. My delivery postal code is {POSTCODE}. Please confirm the estimated delivered price.",
"listing_email_subject": "Enquiry: {LISTING_TITLE} — {CAS_REFERENCE}",
"listing_email_body": "Hello, I am interested in {LISTING_TITLE}, reference {CAS_REFERENCE}. Listing: {PAGE_URL}. Location: {LOCATION}. Price: {PRICE}. Delivery postal code: {POSTCODE}.",
}
# Task 15: French translations for trust interface strings (Canadian French).
TRUST_FR_STRINGS = {
"verified_uk_company": "Concessionnaire canadien vérifié",
"verification_intro": "{company_name} est un concessionnaire canadien inscrit à l'OMVIC. Vous pouvez vérifier nos informations de licence de concessionnaire.",
"verify_companies_house": "Vérifier la licence de concessionnaire",
"company_number": "Numéro de licence de concessionnaire",
"vat_number": "Numéro de TVH",
"registered_office": "Siège social",
"sales_phone": "Téléphone des ventes",
"sales_email": "Courriel des ventes",
"current_machine_location": "Emplacement actuel du véhicule",
"inspection_or_collection": "Inspection ou collecte",
"available_by_appointment": "Disponible sur rendez-vous",
"delivery_insured_europe": "Livraison assurée disponible partout au Canada",
"listing_verification": "Vérification de l'annonce",
"listing_information_verified": "Disponibilité et détails de l'annonce confirmés le : {date}",
"serial_number_status": "Numéro de série",
"inspection_status": "Statut d'inspection",
"condition_check_status": "Vérification de l'état",
"inspection_date": "Date d'inspection",
"availability_checked_today": "Disponibilité vérifiée le : {date}",
"inspection_not_checked": "Non vérifié",
"inspection_pending": "Inspection en attente",
"inspection_visually_checked": "Inspection visuelle effectuée",
"inspection_mechanically_inspected": "Inspection mécanique effectuée",
"inspection_third_party_inspected": "Inspection par un tiers effectuée",
"inspection_documentation_only": "Documentation seulement",
"condition_not_checked": "Non vérifié",
"condition_visually_checked": "Inspection visuelle effectuée",
"condition_mechanically_checked": "Inspection mécanique effectuée",
"condition_third_party_checked": "Vérification par un tiers effectuée",
"condition_documentation_only": "Documentation seulement",
"condition_pending": "En attente",
"collection_available": "Collecte disponible",
"delivery_available": "Livraison disponible",
"yes": "Oui",
"no": "Non",
"vat_short": "TVH appliquée selon votre province d'immatriculation.",
"vat_cross_border_title": "TVH et immatriculation",
"vat_general_guidance": "La TVH est appliquée selon votre province d'immatriculation. Tous les prix sont en dollars canadiens (CAD). Il n'y a pas de frais de douane ou d'importation cachés.",
"vat_business_guidance": "Pour les achats d'entreprise, la TVH peut être récupérable par vos déclarations régulières de crédits de taxe sur les intrants. Nous fournissons une facture appropriée avec notre numéro de TVH.",
"vat_private_guidance": "Pour les achats privés, la TVH applicable et les frais d'immatriculation seront clairement indiqués dans le devis final.",
"vat_final_quote": "Le devis final et la facture confirmeront la TVH, l'immatriculation et les frais de concessionnaire pour la transaction spécifique.",
"vat_disclaimer": "Ces informations sont des conseils généraux et ne remplacent pas un avis fiscal professionnel.",
"company_purchase": "Achat d'entreprise",
"private_purchase": "Achat privé",
"delivery_postcode": "Code postal de livraison",
"country": "Pays",
"preferred_contact_method": "Méthode de contact préférée",
"email_contact": "Courriel",
"telephone_contact": "Téléphone",
"written_support_languages": "Le soutien téléphonique est actuellement disponible en anglais. Les demandes écrites peuvent être traitées dans les langues du site prises en charge via notre système de demande traduit.",
"team": "Équipe",
"recent_deliveries": "Livraisons récentes",
"purchase_process": "Processus d'achat",
"company_gallery": "Galerie de l'entreprise",
"machine_location_not_public": "Emplacement du véhicule disponible sur demande",
"last_checked_inventory": "Dernier inventaire vérifié",
"trust_gallery": "Photos réelles de l'entreprise et de l'entrepôt",
"trust_gallery_intro": "Photographies réelles des locaux, de l'entreposage, du chargement, de l'inspection et des opérations de livraison de {COMPANY_NAME}.",
"buyer_type": "Type d'acheteur",
"listing_whatsapp_message": "Bonjour, je suis intéressé par {LISTING_TITLE}, référence {CAS_REFERENCE}. Mon code postal de livraison est {POSTCODE}. Veuillez confirmer le prix livré estimé.",
"listing_email_subject": "Demande : {LISTING_TITLE} — {CAS_REFERENCE}",
"listing_email_body": "Bonjour, je suis intéressé par {LISTING_TITLE}, référence {CAS_REFERENCE}. Annonce : {PAGE_URL}. Emplacement : {LOCATION}. Prix : {PRICE}. Code postal de livraison : {POSTCODE}.",
}
def now_utc():
return datetime.utcnow().replace(microsecond=0).isoformat()
def _columns(db, table):
return {row[1] for row in db.execute(f"PRAGMA table_info({table})").fetchall()}
def _add_column(db, table, name, definition):
if name not in _columns(db, table):
db.execute(f"ALTER TABLE {table} ADD COLUMN {name} {definition}")
def migrate_trust_schema(db):
"""Additive migration only. Existing rows and global contact settings are preserved."""
settings_columns = {
"company_registration_number": "TEXT",
"vat_number": "TEXT",
"companies_house_url": "TEXT",
"verification_button_label": "TEXT DEFAULT 'Verify dealer license'",
"company_verification_enabled": "INTEGER DEFAULT 0",
"company_verification_listing": "INTEGER DEFAULT 1",
"company_verification_footer": "INTEGER DEFAULT 0",
"team_page_enabled": "INTEGER DEFAULT 0",
"recent_deliveries_page_enabled": "INTEGER DEFAULT 0",
"customer_feedback_enabled": "INTEGER DEFAULT 0",
"listing_whatsapp_template_en": "TEXT",
"listing_email_subject_en": "TEXT",
"listing_email_body_en": "TEXT",
"trust_gallery_home_enabled": "INTEGER DEFAULT 0",
"verification_stale_days": "INTEGER DEFAULT 30",
"show_dealer_license_publicly": "INTEGER DEFAULT 0",
}
for name, definition in settings_columns.items():
_add_column(db, "site_settings", name, definition)
listing_columns = {
"saved_location_id": "INTEGER",
"location_name": "TEXT",
"location_address_line": "TEXT",
"location_postcode": "TEXT",
"location_city": "TEXT",
"location_region": "TEXT",
"location_country": "TEXT",
"location_country_code": "TEXT",
"location_type": "TEXT",
"collection_available": "INTEGER DEFAULT 0",
"inspection_available": "INTEGER DEFAULT 0",
"delivery_available": "INTEGER DEFAULT 1",
"location_notes": "TEXT",
"location_public": "TEXT DEFAULT 'hidden'",
"location_public_text": "TEXT",
"location_updated_at": "TEXT",
"serial_number_status": "TEXT DEFAULT 'pending verification'",
"inspection_status": "TEXT DEFAULT 'not checked'",
"condition_check_status": "TEXT DEFAULT 'not checked'",
"vat_note": "TEXT",
"last_physically_verified_date": "TEXT",
"last_listing_verified_date": "TEXT",
"verification_notes_internal": "TEXT",
"verification_notes_public": "TEXT",
"verified_by_user_id": "INTEGER",
"inspection_date": "TEXT",
}
for name, definition in listing_columns.items():
_add_column(db, "listings", name, definition)
lead_columns = {
"buyer_type": "TEXT",
"delivery_postcode": "TEXT",
"buyer_country": "TEXT",
"preferred_contact_method": "TEXT",
"listing_reference": "TEXT",
"originating_url": "TEXT",
"campaign_source": "TEXT",
}
for name, definition in lead_columns.items():
_add_column(db, "leads", name, definition)
db.executescript(
"""
CREATE TABLE IF NOT EXISTS saved_locations (
id INTEGER PRIMARY KEY,
internal_name TEXT NOT NULL,
public_display_name TEXT,
address_line TEXT,
postcode TEXT,
city TEXT,
region TEXT,
country TEXT,
country_code TEXT,
location_type TEXT DEFAULT 'other',
is_public INTEGER DEFAULT 1,
default_inspection INTEGER DEFAULT 0,
default_collection INTEGER DEFAULT 0,
default_visibility TEXT DEFAULT 'city',
internal_notes TEXT,
is_active INTEGER DEFAULT 1,
created_at TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS company_gallery (
id INTEGER PRIMARY KEY,
local_path TEXT NOT NULL,
category TEXT,
location_id INTEGER,
caption_en TEXT,
alt_text_en TEXT,
sort_order INTEGER DEFAULT 100,
is_published INTEGER DEFAULT 0,
feature_home INTEGER DEFAULT 0,
feature_about INTEGER DEFAULT 0,
feature_warehouse INTEGER DEFAULT 0,
feature_delivery INTEGER DEFAULT 0,
created_at TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS team_members (
id INTEGER PRIMARY KEY,
name TEXT,
job_title_en TEXT,
department TEXT,
photo_path TEXT,
telephone TEXT,
whatsapp TEXT,
email TEXT,
languages_spoken TEXT,
biography_en TEXT,
territory TEXT,
availability TEXT,
sort_order INTEGER DEFAULT 100,
is_active INTEGER DEFAULT 0,
is_public INTEGER DEFAULT 0,
is_placeholder INTEGER DEFAULT 0,
created_at TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS delivery_records (
id INTEGER PRIMARY KEY,
machine_title_en TEXT,
cas_reference TEXT,
destination_country TEXT,
destination_region TEXT,
delivery_month_year TEXT,
delivery_photo_path TEXT,
loading_photo_path TEXT,
description_en TEXT,
customer_name TEXT,
business_name TEXT,
feedback_en TEXT,
is_verified INTEGER DEFAULT 0,
consent_recorded INTEGER DEFAULT 0,
is_published INTEGER DEFAULT 0,
is_placeholder INTEGER DEFAULT 0,
sort_order INTEGER DEFAULT 100,
created_at TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS country_vat_guidance (
id INTEGER PRIMARY KEY,
country TEXT,
language_code TEXT DEFAULT 'en',
business_guidance TEXT,
private_guidance TEXT,
standard_rate TEXT,
special_notes TEXT,
last_reviewed_date TEXT,
legal_source_url TEXT,
is_active INTEGER DEFAULT 0,
created_at TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY,
administrator TEXT,
action TEXT NOT NULL,
entity_type TEXT,
entity_id INTEGER,
old_value TEXT,
new_value TEXT,
created_at TEXT NOT NULL
);
"""
)
# Translation fields for gallery/team/delivery English-source content.
active_codes = [r[0] for r in db.execute("SELECT code FROM languages WHERE is_active=1 ORDER BY sort_order").fetchall()]
for code in active_codes:
if code == "en":
continue
for table, fields in (
("company_gallery", ("caption", "alt_text")),
("team_members", ("job_title", "biography")),
("delivery_records", ("machine_title", "description", "feedback")),
):
for field in fields:
_add_column(db, table, f"{field}_{code}", "TEXT")
# Remove the stale European warehouse location from the old UK site.
db.execute("DELETE FROM saved_locations WHERE internal_name = ?", ("Barleben Warehouse",))
# Known location, intentionally not assigned to any listing.
if not db.execute("SELECT id FROM saved_locations WHERE internal_name=?", ("Ontario Partner Yard",)).fetchone():
db.execute(
"""INSERT INTO saved_locations
(internal_name, public_display_name, postcode, city, country, country_code,
location_type, is_public, default_inspection, default_collection,
default_visibility, is_active, created_at, updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
("Ontario Partner Yard", "Toronto, ON, Canada", "M5H 2N2", "Toronto",
"Canada", "CA", "partner warehouse", 0, 1, 1, "city", 1, now_utc(), now_utc()),
)
# Safe English defaults only; do not touch current company contact data.
db.execute(
"""UPDATE site_settings SET
verification_button_label=COALESCE(NULLIF(verification_button_label,''),'Verify dealer license'),
listing_whatsapp_template_en=COALESCE(NULLIF(listing_whatsapp_template_en,''),'Hello, I am interested in {LISTING_TITLE}, reference {CAS_REFERENCE}. My delivery postal code is {POSTCODE}. Please confirm the estimated delivered price.'),
listing_email_subject_en=COALESCE(NULLIF(listing_email_subject_en,''),'Enquiry: {LISTING_TITLE} — {CAS_REFERENCE}'),
listing_email_body_en=COALESCE(NULLIF(listing_email_body_en,''),'Hello, I am interested in {LISTING_TITLE}, reference {CAS_REFERENCE}. Listing: {PAGE_URL}. Location: {LOCATION}. Price: {PRICE}. Delivery postal code: {POSTCODE}.'),
verification_stale_days=COALESCE(verification_stale_days,30)
WHERE id=1"""
)
# Interface strings are English-source records; translations stay editable/generatable.
interface_cols = _columns(db, "interface_translations") if "interface_translations" in {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} else set()
if interface_cols:
for key, english in TRUST_INTERFACE_STRINGS.items():
if not db.execute("SELECT 1 FROM interface_translations WHERE translation_key=?", (key,)).fetchone():
cols = ["translation_key", "context", "en", "updated_at"]
db.execute(
f"INSERT INTO interface_translations ({','.join(cols)}) VALUES (?,?,?,?)",
(key, "trust_inventory", english, now_utc()),
)
# Force-update existing entries for Canadianized keys so old UK/EU values don't linger
for key, english in TRUST_INTERFACE_STRINGS.items():
db.execute(
"UPDATE interface_translations SET en = ?, updated_at = ? WHERE translation_key = ?",
(english, now_utc(), key),
)
# Also update interface default keys that may have stale UK values in the DB
default_updates = {
"trust_vat_registered": "HST registered",
"trust_uk_based": "Canada based",
"vat_bullet": "OMVIC-registered Canadian dealer.",
"confirm_tax_bullet": "Confirm HST and licensing details case by case.",
"export_support_text": "Canada-wide delivery and paperwork support.",
"export_support_heading": "Canada-Wide Support",
"export_support": "Delivery support",
"footer_company_description": "Used vehicle sales, cars and trucks, and practical Canada-wide support from {COMPANY_NAME}.",
"nav_shipping": "Delivery",
"nav_location": "Our Location",
"chat_vat_q": "Is HST included in the price?",
"trust_europe_delivery": "Canada-wide delivery",
"uk_ireland_europe": "Across Canada",
}
for key, value in default_updates.items():
db.execute(
"UPDATE interface_translations SET en = ?, updated_at = ? WHERE translation_key = ?",
(value, now_utc(), key),
)
# Task 15: Set FR values for trust interface strings where FR column is empty
if "fr" in interface_cols:
for key, french in TRUST_FR_STRINGS.items():
db.execute(
"UPDATE interface_translations SET fr = CASE WHEN COALESCE(fr,'') = '' THEN ? ELSE fr END, updated_at = ? WHERE translation_key = ?",
(french, now_utc(), key),
)
# Also set FR for the default interface keys
default_fr_updates = {
"trust_vat_registered": "Inscrit à la TVH",
"trust_uk_based": "Basé au Canada",
"export_support_text": "Livraison et documentation partout au Canada.",
"export_support_heading": "Soutien partout au Canada",
"export_support": "Soutien à la livraison",
"footer_company_description": "Vente de véhicules d'occasion, voitures et camions, avec soutien partout au Canada par {COMPANY_NAME}.",
"nav_shipping": "Livraison",
"nav_location": "Notre emplacement",
"chat_vat_q": "Les prix incluent-ils la TPS/TVH ?",
"trust_europe_delivery": "Livraison partout au Canada",
"uk_ireland_europe": "Partout au Canada",
}
for key, french in default_fr_updates.items():
db.execute(
"UPDATE interface_translations SET fr = CASE WHEN COALESCE(fr,'') = '' THEN ? ELSE fr END, updated_at = ? WHERE translation_key = ?",
(french, now_utc(), key),
)
# New canonical English pages. Existing pages are never overwritten.
page_defaults = {
"tax-information": (
"HST and Licensing Information",
"General guidance on HST application for vehicle purchases and Canada-wide delivery.",
"HST is applied according to your province
HST is applied according to your province of registration. All pricing is in Canadian dollars (CAD).
For business purchases, HST may be recoverable through your regular input tax credit filings. We provide a proper invoice with our HST number.
For private purchases, applicable HST and licensing fees will be shown clearly in the final quotation.
The final quotation and invoice will confirm HST, licensing, and any dealer fees for the specific transaction.
This information is general guidance and does not replace professional tax advice.
"
),
"purchase-process": (
"Purchase Process and Buyer Information",
"A clear overview of quotation, reservation, inspection, payment, collection and delivery.",
"- Enquire — Contact {COMPANY_NAME} with the vehicle stock number, your name, and delivery postal code. Our team confirms availability and answers your initial questions.
- Confirm the vehicle — Review photos, inspection notes, and available vehicle history documents. Request additional images or a video walk-around if needed.
- Delivered-price estimate — {COMPANY_NAME} provides a written quotation including the vehicle price, applicable HST (based on your province of registration), licensing fees, and delivery cost if required.
- Pickup or Canada-wide delivery — Arrange insured vehicle transport to your location, or collect from the secure storage facility. Delivery options and timelines are confirmed before any commitment is made.
- Financing and HST — Financing is available on approved credit (OAC). HST is applied according to your province of registration. All pricing is in Canadian dollars (CAD).
- Paperwork and handover — Final invoice, ownership transfer documents, safety certification (where applicable), and payment terms are confirmed in writing. Nothing is assumed or implied.
{COMPANY_NAME} is an OMVIC-registered Canadian dealer. The current Terms & Conditions control each transaction.
"
),
}
for slug, (title, excerpt, html) in page_defaults.items():
if not db.execute("SELECT id FROM site_pages WHERE slug=?", (slug,)).fetchone():
db.execute(
"""INSERT INTO site_pages
(slug,title_en,excerpt_en,content_html_en,seo_title_en,meta_description_en,page_type,status,
is_published,show_in_header,show_in_footer,menu_order,english_updated_at,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(slug,title,excerpt,html,title,excerpt,"standard","published",1,0,1,95,now_utc(),now_utc(),now_utc()),
)
def _db():
if "db" not in g:
g.db = sqlite3.connect(current_app.config["DATABASE"])
g.db.row_factory = sqlite3.Row
return g.db
def _actor():
return session.get("username") or f"user:{session.get('user_id','unknown')}"
def audit(action, entity_type="", entity_id=None, old=None, new=None):
_db().execute(
"INSERT INTO audit_log (administrator,action,entity_type,entity_id,old_value,new_value,created_at) VALUES (?,?,?,?,?,?,?)",
(_actor(), action, entity_type, entity_id, json.dumps(old, ensure_ascii=False, default=str) if old is not None else "", json.dumps(new, ensure_ascii=False, default=str) if new is not None else "", now_utc()),
)
def setting(row, key, default=""):
try:
return row[key] if key in row.keys() and row[key] is not None else default
except Exception:
return default
def public_location_text(listing):
mode = setting(listing, "location_public", "hidden") or "hidden"
if mode == "hidden":
return ""
if mode == "custom":
return setting(listing, "location_public_text", "")
country = setting(listing, "location_country", "")
city = setting(listing, "location_city", "")
postcode = setting(listing, "location_postcode", "")
if mode == "country":
return country
if mode == "city":
return ", ".join(part for part in (" ".join(p for p in (postcode, city) if p), country) if part)
address = setting(listing, "location_address_line", "")
region = setting(listing, "location_region", "")
return ", ".join(part for part in (address, " ".join(p for p in (postcode, city) if p), region, country) if part)
def company_verification_data():
settings = _db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
return {
"enabled": bool(setting(settings, "company_verification_enabled", 0)),
"company_name": setting(settings, "company_name"),
"company_registration_number": setting(settings, "company_registration_number"),
"vat_number": setting(settings, "vat_number"),
"address": setting(settings, "address"),
"phone": setting(settings, "phone"),
"whatsapp": setting(settings, "whatsapp"),
"email": setting(settings, "email"),
"companies_house_url": setting(settings, "companies_house_url"),
"button_label": setting(settings, "verification_button_label", "Verify dealer license"),
"show_listing": bool(setting(settings, "company_verification_listing", 1)),
"show_footer": bool(setting(settings, "company_verification_footer", 0)),
"show_dealer_license_publicly": bool(setting(settings, "show_dealer_license_publicly", 0)),
}
def listing_message_context(listing, lang, postcode=""):
settings = _db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
ref = setting(listing, "public_stock_number") or setting(listing, "stock_number")
title = setting(listing, f"title_{lang}") or setting(listing, "title_en")
location = public_location_text(listing) or "Available on request"
price = setting(listing, "price") or setting(listing, "price_label") or "Price on request"
return {
"LISTING_TITLE": title,
"CAS_REFERENCE": ref,
"PAGE_URL": request.url,
"LOCATION": location,
"PRICE": price,
"POSTCODE": postcode,
"LANGUAGE": lang,
}, settings
def replace_placeholders(template, values):
result = template or ""
for key, value in values.items():
result = result.replace("{" + key + "}", str(value or ""))
return result
def translated_interface_value(key, lang, fallback=""):
try:
row = _db().execute(f"SELECT en, {lang} AS localized FROM interface_translations WHERE translation_key=?", (key,)).fetchone()
except Exception:
row = None
if not row:
return fallback
return (row["localized"] or row["en"] or fallback).strip()
def trust_status_label(value, kind="inspection", lang="en"):
slug=re.sub(r"[^a-z0-9]+", "_", (value or "not checked").lower()).strip("_")
key=f"{kind}_{slug}"
return translated_interface_value(key, lang, value or "Not checked")
def listing_whatsapp_template(lang="en"):
settings = _db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
return translated_interface_value("listing_whatsapp_message", lang, setting(settings, "listing_whatsapp_template_en"))
def listing_email_subject_template(lang="en"):
settings = _db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
return translated_interface_value("listing_email_subject", lang, setting(settings, "listing_email_subject_en"))
def listing_email_body_template(lang="en"):
settings = _db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
return translated_interface_value("listing_email_body", lang, setting(settings, "listing_email_body_en"))
def clean_whatsapp_number(value):
return re.sub(r"\D+", "", value or "")
def listing_whatsapp_url(listing, lang="en", postcode=""):
values, settings = listing_message_context(listing, lang, postcode)
template = listing_whatsapp_template(lang)
message = replace_placeholders(template, values)
number = re.sub(r"\D+", "", setting(settings, "whatsapp"))
return f"https://wa.me/{number}?text={quote(message)}" if number else ""
def listing_email_url(listing, lang="en", postcode=""):
values, settings = listing_message_context(listing, lang, postcode)
subject = replace_placeholders(listing_email_subject_template(lang), values)
body = replace_placeholders(listing_email_body_template(lang), values)
email = setting(settings, "email")
return f"mailto:{email}?subject={quote(subject)}&body={quote(body)}" if email else ""
def trust_gallery_images(placement):
field = {"home":"feature_home","about":"feature_about","warehouse":"feature_warehouse","delivery":"feature_delivery"}.get(placement)
if not field:
return []
return _db().execute(f"SELECT * FROM company_gallery WHERE is_published=1 AND {field}=1 ORDER BY sort_order,id").fetchall()
def trust_dashboard_counts():
db = _db()
settings = db.execute("SELECT * FROM site_settings WHERE id=1").fetchone()
stale_days = int(setting(settings, "verification_stale_days", 30) or 30)
cutoff = (date.fromisoformat(TODAY_ISO) - timedelta(days=stale_days)).isoformat()
return {
"verification_url": bool(setting(settings, "companies_house_url")),
"company_number": bool(setting(settings, "company_registration_number")),
"vat_number": bool(setting(settings, "vat_number")),
"phone": bool(setting(settings, "phone")),
"whatsapp": bool(setting(settings, "whatsapp")),
"email": bool(setting(settings, "email")),
"gallery_published": db.execute("SELECT COUNT(*) FROM company_gallery WHERE is_published=1").fetchone()[0],
"locations": db.execute("SELECT COUNT(*) FROM saved_locations WHERE is_active=1").fetchone()[0],
"missing_location": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND saved_location_id IS NULL AND COALESCE(location_country,'')='' ").fetchone()[0],
"missing_verification": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND COALESCE(last_listing_verified_date,'')='' ").fetchone()[0],
"stale_verification": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND last_listing_verified_date!='' AND last_listing_verified_date < ?", (cutoff,)).fetchone()[0],
"hidden_location": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND COALESCE(location_public,'hidden')='hidden'").fetchone()[0],
"missing_refs": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND COALESCE(public_stock_number,'')='' ").fetchone()[0],
"available_unverified": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND status='Available' AND (COALESCE(last_listing_verified_date,'')='' OR last_listing_verified_date < ?)", (cutoff,)).fetchone()[0],
"team_enabled": bool(setting(settings, "team_page_enabled", 0)),
"deliveries_enabled": bool(setting(settings, "recent_deliveries_page_enabled", 0)),
"untranslated_it_listings": db.execute("SELECT COUNT(*) FROM listings WHERE COALESCE(is_deleted,0)=0 AND (COALESCE(title_it,'')='' OR COALESCE(description_html_it,'')='')").fetchone()[0],
"untranslated_contact_strings": db.execute("SELECT COUNT(*) FROM interface_translations WHERE translation_key IN ('name','email','phone','message','send_message','company_purchase','private_purchase','delivery_postcode','preferred_contact_method') AND COALESCE(it,'')=''").fetchone()[0],
"untranslated_whatsapp_templates": db.execute("SELECT COUNT(*) FROM interface_translations WHERE translation_key IN ('listing_whatsapp_message','listing_email_subject','listing_email_body') AND COALESCE(it,'')=''").fetchone()[0],
}
def _listing_ids_from_request():
# "All matching current filters" deliberately overrides visible row selections.
# This makes the scope explicit and avoids silently updating only the current page.
if request.form.get("all_matching") != "1":
return [int(v) for v in request.form.getlist("listing_ids") if str(v).isdigit()]
clauses = ["COALESCE(is_deleted,0)=0"]
params = []
mapping = {"category":"category","make":"make","model":"model","status":"status","year":"year"}
for form_key, column in mapping.items():
value = request.form.get(form_key, "").strip()
if value:
clauses.append(f"{column}=?")
params.append(value)
q = request.form.get("q", "").strip()
if q:
clauses.append("(title_en LIKE ? OR stock_number LIKE ? OR public_stock_number LIKE ? OR make LIKE ? OR model LIKE ?)")
params.extend([f"%{q}%"]*5)
return [r[0] for r in _db().execute(f"SELECT id FROM listings WHERE {' AND '.join(clauses)}", params).fetchall()]
def create_blueprint(login_required, admin_prefix):
bp = Blueprint("trust", __name__)
@bp.route(f"{admin_prefix}/trust", methods=("GET", "POST"))
@login_required
def admin_trust():
db = _db()
settings = db.execute("SELECT * FROM site_settings WHERE id=1").fetchone()
if request.method == "POST":
old = dict(settings)
fields = (
"company_registration_number","vat_number","companies_house_url","verification_button_label",
"listing_whatsapp_template_en","listing_email_subject_en","listing_email_body_en","verification_stale_days"
)
values = [request.form.get(f, "").strip() for f in fields]
toggles = {
"company_verification_enabled": 1 if request.form.get("company_verification_enabled") else 0,
"company_verification_listing": 1 if request.form.get("company_verification_listing") else 0,
"company_verification_footer": 1 if request.form.get("company_verification_footer") else 0,
"team_page_enabled": 1 if request.form.get("team_page_enabled") else 0,
"recent_deliveries_page_enabled": 1 if request.form.get("recent_deliveries_page_enabled") else 0,
"customer_feedback_enabled": 1 if request.form.get("customer_feedback_enabled") else 0,
"trust_gallery_home_enabled": 1 if request.form.get("trust_gallery_home_enabled") else 0,
"show_dealer_license_publicly": 1 if request.form.get("show_dealer_license_publicly") else 0,
}
assignments = [f"{f}=?" for f in fields] + [f"{k}=?" for k in toggles]
db.execute(f"UPDATE site_settings SET {','.join(assignments)},updated_at=? WHERE id=1", (*values,*toggles.values(),now_utc()))
english_updates = {
"verify_companies_house": request.form.get("verification_button_label", "").strip() or "Verify dealer license",
"listing_whatsapp_message": request.form.get("listing_whatsapp_template_en", "").strip(),
"listing_email_subject": request.form.get("listing_email_subject_en", "").strip(),
"listing_email_body": request.form.get("listing_email_body_en", "").strip(),
}
for translation_key, english_value in english_updates.items():
db.execute("UPDATE interface_translations SET en=?, updated_at=? WHERE translation_key=?", (english_value, now_utc(), translation_key))
audit("company/trust settings changed", "site_settings", 1, old, {**dict(zip(fields,values)),**toggles})
db.commit()
flash("Trust settings updated. Existing contact details were preserved.", "success")
return redirect(url_for("trust.admin_trust"))
return render_template("admin/trust_dashboard.html", settings=settings, counts=trust_dashboard_counts(), today=TODAY_ISO)
@bp.route(f"{admin_prefix}/locations", methods=("GET", "POST"))
@login_required
def admin_locations():
db = _db()
if request.method == "POST":
if request.form.get("location_action") == "bulk_import":
raw=request.form.get("bulk_rows", "").strip()
created=0; errors=[]
for line_no,line in enumerate(raw.splitlines(), start=1):
if not line.strip(): continue
parts=[part.strip() for part in re.split(r"\t|\||;", line)]
if line_no==1 and parts and parts[0].lower() in ("internal name","internal_name"): continue
parts += [""]*(10-len(parts))
internal,public,address,postcode,city,region,country,code,loc_type,visibility=parts[:10]
if not internal:
errors.append(f"row {line_no}: missing internal name"); continue
db.execute("""INSERT INTO saved_locations (internal_name,public_display_name,address_line,postcode,city,region,country,country_code,location_type,default_visibility,is_public,default_inspection,default_collection,is_active,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(internal,public,address,postcode,city,region,country,code.upper(),loc_type or "other",visibility or "city",1,0,0,1,now_utc(),now_utc()))
created+=1
audit("saved locations bulk imported","saved_location",None,None,{"created":created,"errors":errors[:10]}); db.commit()
flash(f"Imported {created} saved location{'s' if created!=1 else ''}.", "success")
if errors: flash("Some rows were skipped: " + " | ".join(errors[:5]), "warning")
return redirect(url_for("trust.admin_locations"))
loc_id = request.form.get("location_id", "")
fields = ("internal_name","public_display_name","address_line","postcode","city","region","country","country_code","location_type","default_visibility","internal_notes")
values = [request.form.get(f, "").strip() for f in fields]
bools = [1 if request.form.get(x) else 0 for x in ("is_public","default_inspection","default_collection","is_active")]
if loc_id.isdigit():
old = db.execute("SELECT * FROM saved_locations WHERE id=?", (int(loc_id),)).fetchone()
db.execute(f"UPDATE saved_locations SET {','.join(f'{f}=?' for f in fields)},is_public=?,default_inspection=?,default_collection=?,is_active=?,updated_at=? WHERE id=?", (*values,*bools,now_utc(),int(loc_id)))
audit("saved warehouse changed", "saved_location", int(loc_id), dict(old) if old else None, dict(zip(fields,values)))
else:
cur = db.execute(f"INSERT INTO saved_locations ({','.join(fields)},is_public,default_inspection,default_collection,is_active,created_at,updated_at) VALUES ({','.join('?' for _ in range(len(fields)+6))})", (*values,*bools,now_utc(),now_utc()))
audit("saved warehouse created", "saved_location", cur.lastrowid, None, dict(zip(fields,values)))
db.commit(); flash("Location saved.", "success")
return redirect(url_for("trust.admin_locations"))
locations = db.execute("SELECT * FROM saved_locations ORDER BY is_active DESC,internal_name").fetchall()
return render_template("admin/locations.html", locations=locations, location_types=LOCATION_TYPES, visibility_modes=LOCATION_VISIBILITY)
@bp.route(f"{admin_prefix}/listings//trust", methods=("POST",))
@login_required
def admin_listing_trust(listing_id):
db = _db(); listing = db.execute("SELECT * FROM listings WHERE id=?", (listing_id,)).fetchone()
if not listing: abort(404)
old = dict(listing)
location_id = request.form.get("saved_location_id", "")
values = {
"saved_location_id": int(location_id) if location_id.isdigit() else None,
"location_name": request.form.get("location_name", "").strip(),
"location_address_line": request.form.get("location_address_line", "").strip(),
"location_postcode": request.form.get("location_postcode", "").strip(),
"location_city": request.form.get("location_city", "").strip(),
"location_region": request.form.get("location_region", "").strip(),
"location_country": request.form.get("location_country", "").strip(),
"location_country_code": request.form.get("location_country_code", "").strip().upper(),
"location_type": request.form.get("location_type", "other").strip(),
"collection_available": 1 if request.form.get("collection_available") else 0,
"inspection_available": 1 if request.form.get("inspection_available") else 0,
"delivery_available": 1 if request.form.get("delivery_available") else 0,
"location_notes": request.form.get("location_notes", "").strip(),
"location_public": request.form.get("location_public", "hidden").strip(),
"location_public_text": request.form.get("location_public_text", "").strip(),
"location_updated_at": request.form.get("location_updated_at", "").strip() or None,
"serial_number_status": request.form.get("serial_number_status", "pending verification"),
"inspection_status": request.form.get("inspection_status", "not checked"),
"inspection_date": request.form.get("inspection_date", "").strip() or None,
"condition_check_status": request.form.get("condition_check_status", "not checked"),
"vat_note": request.form.get("vat_note", "").strip(),
"last_physically_verified_date": request.form.get("last_physically_verified_date", "").strip() or None,
"last_listing_verified_date": request.form.get("last_listing_verified_date", "").strip() or None,
"verification_notes_internal": request.form.get("verification_notes_internal", "").strip(),
"verification_notes_public": request.form.get("verification_notes_public", "").strip(),
"verified_by_user_id": session.get("user_id") if request.form.get("last_listing_verified_date") else setting(listing,"verified_by_user_id",None),
}
if values["saved_location_id"]:
loc = db.execute("SELECT * FROM saved_locations WHERE id=?", (values["saved_location_id"],)).fetchone()
if loc:
values.update({
"location_name": loc["public_display_name"] or loc["internal_name"], "location_address_line": loc["address_line"],
"location_postcode": loc["postcode"], "location_city": loc["city"], "location_region": loc["region"],
"location_country": loc["country"], "location_country_code": loc["country_code"], "location_type": loc["location_type"],
})
db.execute(f"UPDATE listings SET {','.join(f'{k}=?' for k in values)},updated_at=? WHERE id=?", (*values.values(),now_utc(),listing_id))
audit("listing trust/location changed", "listing", listing_id, old, values)
db.commit(); flash("Listing location and verification updated.", "success")
return redirect(url_for("admin_listing_edit", id=listing_id))
@bp.route(f"{admin_prefix}/listings/trust-bulk", methods=("POST",))
@login_required
def admin_listing_trust_bulk():
db = _db(); ids = _listing_ids_from_request()
if not ids:
flash("Select listings or choose all matching filters.", "warning"); return redirect(url_for("admin_listings"))
action = request.form.get("trust_action", "")
placeholders = ",".join("?" for _ in ids); now=now_utc()
if action == "assign_location":
loc_id = request.form.get("saved_location_id", "")
loc = db.execute("SELECT * FROM saved_locations WHERE id=?", (int(loc_id),)).fetchone() if loc_id.isdigit() else None
if not loc: flash("Choose a saved location.", "warning"); return redirect(url_for("admin_listings"))
db.execute(f"""UPDATE listings SET saved_location_id=?,location_name=?,location_address_line=?,location_postcode=?,location_city=?,location_region=?,location_country=?,location_country_code=?,location_type=?,collection_available=?,inspection_available=?,location_public=?,location_updated_at=?,updated_at=? WHERE id IN ({placeholders})""",
(loc["id"],loc["public_display_name"] or loc["internal_name"],loc["address_line"],loc["postcode"],loc["city"],loc["region"],loc["country"],loc["country_code"],loc["location_type"],loc["default_collection"],loc["default_inspection"],loc["default_visibility"],TODAY_ISO,now,*ids))
audit("bulk location assignment", "listing", None, None, {"ids":ids,"location_id":loc["id"]})
elif action == "mark_verified_today":
db.execute(f"UPDATE listings SET last_listing_verified_date=?,verified_by_user_id=?,updated_at=? WHERE id IN ({placeholders})", (TODAY_ISO,session.get("user_id"),now,*ids))
audit("selected listings verified today", "listing", None, None, {"ids":ids,"date":TODAY_ISO})
elif action == "set_verification_date":
verified_date=request.form.get("verification_date", "").strip()
db.execute(f"UPDATE listings SET last_listing_verified_date=?,verified_by_user_id=?,updated_at=? WHERE id IN ({placeholders})", (verified_date or None,session.get("user_id") if verified_date else None,now,*ids))
audit("bulk verification date updated", "listing", None, None, {"ids":ids,"date":verified_date})
elif action == "clear_verification_date":
db.execute(f"UPDATE listings SET last_listing_verified_date=NULL,verified_by_user_id=NULL,updated_at=? WHERE id IN ({placeholders})", (now,*ids))
audit("bulk verification date cleared", "listing", None, None, {"ids":ids})
elif action == "set_visibility":
mode=request.form.get("location_public", "hidden")
db.execute(f"UPDATE listings SET location_public=?,updated_at=? WHERE id IN ({placeholders})", (mode,now,*ids))
audit("bulk public location visibility changed", "listing", None, None, {"ids":ids,"mode":mode})
elif action in ("collection_yes","collection_no","inspection_yes","inspection_no"):
field="collection_available" if action.startswith("collection") else "inspection_available"; val=1 if action.endswith("yes") else 0
db.execute(f"UPDATE listings SET {field}=?,updated_at=? WHERE id IN ({placeholders})", (val,now,*ids))
audit(f"bulk {field} changed", "listing", None, None, {"ids":ids,"value":val})
elif action == "randomize_inspection_dates":
start_text=request.form.get("inspection_date_from", "").strip()
end_text=request.form.get("inspection_date_to", "").strip()
try:
start_date=date.fromisoformat(start_text); end_date=date.fromisoformat(end_text)
if end_date < start_date: start_date,end_date=end_date,start_date
except ValueError:
flash("Choose a valid inspection date range.", "warning"); return redirect(url_for("admin_listings"))
status=request.form.get("bulk_inspection_status", "mechanically inspected").strip()
requested_condition=request.form.get("bulk_condition_check_status", "auto").strip()
condition_map={
"visually checked": "visually checked",
"mechanically inspected": "mechanically checked",
"third-party inspected": "third-party checked",
"documentation only": "documentation only",
}
condition_status = condition_map.get(status, "not checked") if requested_condition == "auto" else requested_condition
span=(end_date-start_date).days
for listing_id in ids:
chosen=start_date+timedelta(days=random.randint(0, span if span>0 else 0))
db.execute("UPDATE listings SET inspection_date=?,inspection_status=?,condition_check_status=?,inspection_available=1,updated_at=? WHERE id=?", (chosen.isoformat(),status,condition_status,now,listing_id))
audit("bulk inspection dates randomized", "listing", None, None, {"ids":ids,"from":start_text,"to":end_text,"status":status,"condition_status":condition_status})
elif action == "update_notes":
notes=request.form.get("location_notes", "").strip()
db.execute(f"UPDATE listings SET location_notes=?,updated_at=? WHERE id IN ({placeholders})", (notes,now,*ids))
audit("bulk location notes changed", "listing", None, None, {"ids":ids,"notes":notes})
else:
flash("Choose a valid trust/inventory action.", "warning"); return redirect(url_for("admin_listings"))
db.commit()
action_label = action.replace("_", " ").title()
flash(f"{action_label}: updated {len(ids)} listing{'s' if len(ids)!=1 else ''}. Review the Location / Inspection column below or open any listing to edit the values.", "success")
return redirect(url_for("admin_listings", **{key: request.form.get(key, "") for key in ("q","category","make","model","status","published","year") if request.form.get(key, "")}))
@bp.route(f"{admin_prefix}/company-gallery", methods=("GET","POST"))
@login_required
def admin_company_gallery():
db=_db()
if request.method=="POST":
if request.form.get("gallery_action") == "bulk_update":
ids=[int(v) for v in request.form.getlist("image_ids") if v.isdigit()]
if not ids:
flash("Select at least one gallery image.","warning"); return redirect(url_for("trust.admin_company_gallery"))
updates=[]; params=[]
if request.form.get("bulk_category"):
updates.append("category=?"); params.append(request.form.get("bulk_category"))
if request.form.get("bulk_location_id"):
updates.append("location_id=?"); params.append(int(request.form.get("bulk_location_id")))
for field in ("is_published","feature_home","feature_about","feature_warehouse","feature_delivery"):
mode=request.form.get("bulk_"+field,"")
if mode in ("0","1"):
updates.append(f"{field}=?"); params.append(int(mode))
if updates:
marks=','.join('?' for _ in ids)
db.execute(f"UPDATE company_gallery SET {','.join(updates)},updated_at=? WHERE id IN ({marks})", (*params,now_utc(),*ids))
audit("company gallery bulk updated","company_gallery",None,None,{"ids":ids,"fields":updates})
db.commit()
flash(f"Updated {len(ids)} gallery image{'s' if len(ids)!=1 else ''}.","success")
return redirect(url_for("trust.admin_company_gallery"))
uploads=[f for f in request.files.getlist("images") if f and f.filename]
single=request.files.get("image")
if single and single.filename: uploads.append(single)
if not uploads:
flash("Choose one or more real company images.","warning"); return redirect(url_for("trust.admin_company_gallery"))
folder=Path(current_app.root_path)/"static/uploads/company_gallery"; folder.mkdir(parents=True,exist_ok=True)
created=0
for index,uploaded in enumerate(uploads, start=1):
base_name=secure_filename(Path(uploaded.filename).stem.lower().replace(" ","-"))[:80] or "company-photo"
target=folder/f"clean-agri-sales-{base_name}-{datetime.utcnow().strftime('%Y%m%d%H%M%S%f')}.webp"
if Image is not None and ImageOps is not None:
image=ImageOps.exif_transpose(Image.open(uploaded.stream)).convert("RGB"); image.thumbnail((1800,1350)); image.save(target,"WEBP",quality=82,method=6)
thumb=target.with_name(target.stem+"_thumb.webp"); ti=image.copy(); ti.thumbnail((700,525)); ti.save(thumb,"WEBP",quality=72,method=6)
else:
target=folder/f"{datetime.utcnow().strftime('%Y%m%d%H%M%S%f')}-{secure_filename(uploaded.filename)}"; uploaded.save(target)
local_path="/static/uploads/company_gallery/"+target.name
cur=db.execute("""INSERT INTO company_gallery (local_path,category,location_id,caption_en,alt_text_en,sort_order,is_published,feature_home,feature_about,feature_warehouse,feature_delivery,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(local_path,request.form.get("category","company premises"),int(request.form["location_id"]) if request.form.get("location_id","").isdigit() else None,request.form.get("caption_en","").strip(),request.form.get("alt_text_en","").strip(),int(request.form.get("sort_order") or 100)+index-1,1 if request.form.get("is_published") else 0,1 if request.form.get("feature_home") else 0,1 if request.form.get("feature_about") else 0,1 if request.form.get("feature_warehouse") else 0,1 if request.form.get("feature_delivery") else 0,now_utc(),now_utc()))
audit("company gallery image uploaded","company_gallery",cur.lastrowid,None,{"path":local_path,"published":bool(request.form.get("is_published"))}); created+=1
db.commit(); flash(f"Uploaded {created} company image{'s' if created!=1 else ''}. New images remain unpublished unless Publish was checked.","success"); return redirect(url_for("trust.admin_company_gallery"))
rows=db.execute("SELECT g.*,l.internal_name location_internal_name FROM company_gallery g LEFT JOIN saved_locations l ON l.id=g.location_id ORDER BY g.sort_order,g.id").fetchall()
locations=db.execute("SELECT * FROM saved_locations WHERE is_active=1 ORDER BY internal_name").fetchall()
return render_template("admin/company_gallery.html", images=rows, locations=locations, categories=GALLERY_CATEGORIES)
@bp.route(f"{admin_prefix}/company-gallery//update", methods=("POST",))
@login_required
def admin_company_gallery_update(image_id):
db=_db(); row=db.execute("SELECT * FROM company_gallery WHERE id=?",(image_id,)).fetchone()
if not row: abort(404)
values={
"category":request.form.get("category",row["category"] or "company premises"),
"location_id":int(request.form["location_id"]) if request.form.get("location_id","").isdigit() else None,
"caption_en":request.form.get("caption_en","").strip(),
"alt_text_en":request.form.get("alt_text_en","").strip(),
"sort_order":int(request.form.get("sort_order") or 100),
"is_published":1 if request.form.get("is_published") else 0,
"feature_home":1 if request.form.get("feature_home") else 0,
"feature_about":1 if request.form.get("feature_about") else 0,
"feature_warehouse":1 if request.form.get("feature_warehouse") else 0,
"feature_delivery":1 if request.form.get("feature_delivery") else 0,
}
db.execute(f"UPDATE company_gallery SET {','.join(f'{k}=?' for k in values)},updated_at=? WHERE id=?",(*values.values(),now_utc(),image_id))
audit("company gallery image updated","company_gallery",image_id,dict(row),values); db.commit(); flash("Company gallery image updated.","success")
return redirect(url_for("trust.admin_company_gallery"))
@bp.route(f"{admin_prefix}/company-gallery//toggle", methods=("POST",))
@login_required
def admin_company_gallery_toggle(image_id):
db=_db(); row=db.execute("SELECT * FROM company_gallery WHERE id=?",(image_id,)).fetchone()
if not row: abort(404)
val=0 if row["is_published"] else 1; db.execute("UPDATE company_gallery SET is_published=?,updated_at=? WHERE id=?",(val,now_utc(),image_id)); audit("company gallery publish changed","company_gallery",image_id,{"published":row["is_published"]},{"published":val}); db.commit(); return redirect(url_for("trust.admin_company_gallery"))
@bp.route(f"{admin_prefix}/team", methods=("GET","POST"))
@login_required
def admin_team():
db=_db()
if request.method=="POST":
cur=db.execute("""INSERT INTO team_members (name,job_title_en,department,telephone,whatsapp,email,languages_spoken,biography_en,territory,availability,sort_order,is_active,is_public,is_placeholder,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(request.form.get("name","").strip() or "PLACEHOLDER — REPLACE BEFORE PUBLISHING",request.form.get("job_title_en","").strip(),request.form.get("department","").strip(),request.form.get("telephone","").strip(),request.form.get("whatsapp","").strip(),request.form.get("email","").strip(),request.form.get("languages_spoken","").strip(),request.form.get("biography_en","").strip(),request.form.get("territory","").strip(),request.form.get("availability","").strip(),int(request.form.get("sort_order") or 100),0,0,1,now_utc(),now_utc()))
audit("placeholder team record created", "team_member", cur.lastrowid, None, {"public":False}); db.commit(); flash("Placeholder created unpublished. Replace and explicitly publish later.","success"); return redirect(url_for("trust.admin_team"))
return render_template("admin/team.html", members=db.execute("SELECT * FROM team_members ORDER BY sort_order,id").fetchall(), settings=db.execute("SELECT * FROM site_settings WHERE id=1").fetchone())
@bp.route(f"{admin_prefix}/team//update", methods=("POST",))
@login_required
def admin_team_update(member_id):
db=_db(); row=db.execute("SELECT * FROM team_members WHERE id=?",(member_id,)).fetchone()
if not row: abort(404)
photo_path=row["photo_path"] or ""; uploaded=request.files.get("photo")
if uploaded and uploaded.filename:
folder=Path(current_app.root_path)/"static/uploads/team"; folder.mkdir(parents=True,exist_ok=True)
target=folder/f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}-{secure_filename(uploaded.filename)}"; uploaded.save(target); photo_path="/static/uploads/team/"+target.name
is_placeholder=0 if request.form.get("confirmed_real_staff") else 1
requested_public=1 if request.form.get("is_public") else 0
is_active=1 if request.form.get("is_active") else 0
if requested_public and (is_placeholder or not request.form.get("name","").strip() or not request.form.get("job_title_en","").strip()):
flash("A public team member must be confirmed real staff with a real name and job title.","warning"); return redirect(url_for("trust.admin_team"))
values={"name":request.form.get("name","").strip(),"job_title_en":request.form.get("job_title_en","").strip(),"department":request.form.get("department","").strip(),"photo_path":photo_path,"telephone":request.form.get("telephone","").strip(),"whatsapp":request.form.get("whatsapp","").strip(),"email":request.form.get("email","").strip(),"languages_spoken":request.form.get("languages_spoken","").strip(),"biography_en":request.form.get("biography_en","").strip(),"territory":request.form.get("territory","").strip(),"availability":request.form.get("availability","").strip(),"sort_order":int(request.form.get("sort_order") or 100),"is_active":is_active,"is_public":requested_public,"is_placeholder":is_placeholder}
db.execute(f"UPDATE team_members SET {','.join(f'{k}=?' for k in values)},updated_at=? WHERE id=?",(*values.values(),now_utc(),member_id)); audit("staff member updated/published" if requested_public else "staff member updated","team_member",member_id,dict(row),values); db.commit(); flash("Team member updated.","success"); return redirect(url_for("trust.admin_team"))
@bp.route(f"{admin_prefix}/deliveries", methods=("GET","POST"))
@login_required
def admin_deliveries():
db=_db()
if request.method=="POST":
cur=db.execute("""INSERT INTO delivery_records (machine_title_en,cas_reference,destination_country,destination_region,delivery_month_year,description_en,feedback_en,is_verified,consent_recorded,is_published,is_placeholder,sort_order,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(request.form.get("machine_title_en","").strip() or "PLACEHOLDER DELIVERY RECORD — REPLACE WITH VERIFIED CUSTOMER DELIVERY.",request.form.get("cas_reference","").strip(),request.form.get("destination_country","").strip(),request.form.get("destination_region","").strip(),request.form.get("delivery_month_year","").strip(),request.form.get("description_en","").strip(),request.form.get("feedback_en","").strip(),0,0,0,1,int(request.form.get("sort_order") or 100),now_utc(),now_utc()))
audit("placeholder delivery record created","delivery_record",cur.lastrowid,None,{"public":False}); db.commit(); flash("Placeholder delivery record created unpublished.","success"); return redirect(url_for("trust.admin_deliveries"))
return render_template("admin/deliveries.html", records=db.execute("SELECT * FROM delivery_records ORDER BY sort_order,id").fetchall(), settings=db.execute("SELECT * FROM site_settings WHERE id=1").fetchone())
@bp.route(f"{admin_prefix}/deliveries//update", methods=("POST",))
@login_required
def admin_delivery_update(record_id):
db=_db(); row=db.execute("SELECT * FROM delivery_records WHERE id=?",(record_id,)).fetchone()
if not row: abort(404)
paths={"delivery_photo_path":row["delivery_photo_path"] or "","loading_photo_path":row["loading_photo_path"] or ""}
folder=Path(current_app.root_path)/"static/uploads/deliveries"; folder.mkdir(parents=True,exist_ok=True)
for field,input_name in (("delivery_photo_path","delivery_photo"),("loading_photo_path","loading_photo")):
uploaded=request.files.get(input_name)
if uploaded and uploaded.filename:
target=folder/f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}-{secure_filename(uploaded.filename)}"; uploaded.save(target); paths[field]="/static/uploads/deliveries/"+target.name
is_placeholder=0 if request.form.get("confirmed_real_delivery") else 1
verified=1 if request.form.get("is_verified") else 0; consent=1 if request.form.get("consent_recorded") else 0; publish=1 if request.form.get("is_published") else 0
if publish and (is_placeholder or not verified or not consent):
flash("Public delivery records must be confirmed real, verified, and have consent recorded.","warning"); return redirect(url_for("trust.admin_deliveries"))
values={"machine_title_en":request.form.get("machine_title_en","").strip(),"cas_reference":request.form.get("cas_reference","").strip(),"destination_country":request.form.get("destination_country","").strip(),"destination_region":request.form.get("destination_region","").strip(),"delivery_month_year":request.form.get("delivery_month_year","").strip(),**paths,"description_en":request.form.get("description_en","").strip(),"customer_name":request.form.get("customer_name","").strip(),"business_name":request.form.get("business_name","").strip(),"feedback_en":request.form.get("feedback_en","").strip(),"is_verified":verified,"consent_recorded":consent,"is_published":publish,"is_placeholder":is_placeholder,"sort_order":int(request.form.get("sort_order") or 100)}
db.execute(f"UPDATE delivery_records SET {','.join(f'{k}=?' for k in values)},updated_at=? WHERE id=?",(*values.values(),now_utc(),record_id)); audit("customer feedback/delivery published" if publish else "delivery record updated","delivery_record",record_id,dict(row),values); db.commit(); flash("Delivery record updated.","success"); return redirect(url_for("trust.admin_deliveries"))
@bp.route(f"{admin_prefix}/audit-log")
@login_required
def admin_audit_log():
return render_template("admin/audit_log.html", rows=_db().execute("SELECT * FROM audit_log ORDER BY id DESC LIMIT 500").fetchall())
@bp.route("//team")
def team_page(lang):
settings=_db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
if not setting(settings,"team_page_enabled",0): abort(404)
members=_db().execute("SELECT * FROM team_members WHERE is_active=1 AND is_public=1 AND is_placeholder=0 ORDER BY sort_order,id").fetchall()
return render_template("team.html", members=members, lang=lang)
@bp.route("//recent-deliveries")
def recent_deliveries_page(lang):
settings=_db().execute("SELECT * FROM site_settings WHERE id=1").fetchone()
if not setting(settings,"recent_deliveries_page_enabled",0): abort(404)
rows=_db().execute("SELECT * FROM delivery_records WHERE is_published=1 AND is_verified=1 AND consent_recorded=1 AND is_placeholder=0 ORDER BY sort_order,id").fetchall()
return render_template("recent_deliveries.html", records=rows, lang=lang)
@bp.route(f"{admin_prefix}/team/preview")
@login_required
def admin_team_preview():
members=_db().execute("SELECT * FROM team_members ORDER BY sort_order,id").fetchall(); return render_template("team.html",members=members,lang=request.args.get("lang","en"),admin_preview=True)
@bp.route(f"{admin_prefix}/deliveries/preview")
@login_required
def admin_deliveries_preview():
rows=_db().execute("SELECT * FROM delivery_records ORDER BY sort_order,id").fetchall(); return render_template("recent_deliveries.html",records=rows,lang=request.args.get("lang","en"),admin_preview=True)
return bp
def format_canadian_date(value):
"""Format ISO YYYY-MM-DD values as YYYY-MM-DD for public/admin display."""
if not value:
return ""
text = str(value).strip()
try:
date.fromisoformat(text[:10])
return text[:10]
except (TypeError, ValueError):
return text
def format_address(settings):
"""Return a 2-line HTML address: Street
City, Province Postal.
Falls back to the legacy single-line `address` field when the 4
structured fields are all empty. Returns safe HTML (caller uses |safe)."""
from markupsafe import escape
def g(key):
try:
return (settings.get(key) or "").strip()
except AttributeError:
return (settings[key] or "").strip() if key in settings.keys() else ""
street = str(escape(g("address_street")))
city = str(escape(g("address_city")))
province = str(escape(g("address_province")))
postal = str(escape(g("address_postal")))
if not (street or city or province or postal):
return str(escape(g("address")))
line2 = ", ".join(p for p in (city, (province + " " + postal).strip()) if p)
if street and line2:
return f"{street}
{line2}"
return street or line2
def register_trust_features(app, login_required, admin_prefix):
bp=create_blueprint(login_required, admin_prefix); app.register_blueprint(bp)
app.jinja_env.globals.update(
public_location_text=public_location_text,
company_verification_data=company_verification_data,
listing_whatsapp_url=listing_whatsapp_url,
listing_email_url=listing_email_url,
listing_whatsapp_template=listing_whatsapp_template,
listing_email_subject_template=listing_email_subject_template,
listing_email_body_template=listing_email_body_template,
clean_whatsapp_number=clean_whatsapp_number,
trust_gallery_images=trust_gallery_images,
trust_status_label=trust_status_label,
trust_today=TODAY_ISO,
format_canadian_date=format_canadian_date,
format_address=format_address,
)