90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""Homepage showcase configuration and URL validation."""
|
|
|
|
|
|
from database_service import database_connection
|
|
import json
|
|
from runtime_config import (
|
|
DATABASE_URL,
|
|
logger,
|
|
)
|
|
import urllib.parse
|
|
|
|
__all__ = ["get_homepage_showcase_links", "save_homepage_showcase_links"]
|
|
|
|
HOMEPAGE_SHOWCASE_KEY = "homepage_showcase"
|
|
|
|
|
|
def _default_homepage_showcase_links():
|
|
return [
|
|
{"label": "Site 1", "url": ""},
|
|
{"label": "Site 2", "url": ""},
|
|
{"label": "Site 3", "url": ""},
|
|
]
|
|
|
|
def get_homepage_showcase_links():
|
|
"""Load 3 homepage showcase slots from site_settings."""
|
|
default = _default_homepage_showcase_links()
|
|
try:
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT value FROM site_settings WHERE key = ?", (HOMEPAGE_SHOWCASE_KEY,))
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
return default
|
|
raw = row['value'] if isinstance(row, dict) else row[0]
|
|
data = json.loads(raw)
|
|
if not isinstance(data, list):
|
|
return default
|
|
out = []
|
|
for i in range(3):
|
|
if i < len(data) and isinstance(data[i], dict):
|
|
label = (data[i].get('label') or f'Site {i + 1}').strip()[:160]
|
|
url = (data[i].get('url') or '').strip()[:2048]
|
|
out.append({"label": label, "url": url})
|
|
else:
|
|
out.append({"label": f'Site {i + 1}', "url": ""})
|
|
return out
|
|
except Exception as e:
|
|
logger.error(f"get_homepage_showcase_links: {e}")
|
|
return default
|
|
|
|
def _normalize_showcase_url(url: str) -> str:
|
|
u = (url or '').strip()
|
|
if not u:
|
|
return ''
|
|
if not u.startswith(('http://', 'https://')):
|
|
u = 'https://' + u
|
|
return u
|
|
|
|
def _is_valid_http_url(u: str) -> bool:
|
|
if not u:
|
|
return True
|
|
try:
|
|
p = urllib.parse.urlparse(u)
|
|
return p.scheme in ('http', 'https') and bool(p.netloc)
|
|
except Exception:
|
|
return False
|
|
|
|
def save_homepage_showcase_links(links: list) -> None:
|
|
payload = json.dumps(links, ensure_ascii=False)
|
|
with database_connection() as conn:
|
|
cursor = conn.cursor()
|
|
if 'postgres' in DATABASE_URL:
|
|
cursor.execute(
|
|
"INSERT INTO site_settings (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = CURRENT_TIMESTAMP",
|
|
(HOMEPAGE_SHOWCASE_KEY, payload)
|
|
)
|
|
else:
|
|
cursor.execute("SELECT 1 FROM site_settings WHERE key = ?", (HOMEPAGE_SHOWCASE_KEY,))
|
|
if cursor.fetchone():
|
|
cursor.execute(
|
|
"UPDATE site_settings SET value = ?, updated_at = CURRENT_TIMESTAMP WHERE key = ?",
|
|
(payload, HOMEPAGE_SHOWCASE_KEY),
|
|
)
|
|
else:
|
|
cursor.execute(
|
|
"INSERT INTO site_settings (key, value) VALUES (?, ?)",
|
|
(HOMEPAGE_SHOWCASE_KEY, payload),
|
|
)
|
|
conn.commit()
|