105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# Ensure project root is on PYTHONPATH when running from scripts/
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
if project_root not in sys.path:
|
||
sys.path.insert(0, project_root)
|
||
|
||
from app_core.config.settings import AppSettings
|
||
from app_core.db.database import engine
|
||
from sqlalchemy import text
|
||
|
||
|
||
def check_env(settings: AppSettings) -> list[str]:
|
||
missing: list[str] = []
|
||
required = [
|
||
("DATABASE_URL", settings.database_url or os.getenv("DATABASE_URL")),
|
||
("SMTP_HOST", settings.smtp_host),
|
||
("SMTP_PORT", settings.smtp_port),
|
||
("SMTP_USER", settings.smtp_user),
|
||
("SMTP_PASSWORD", settings.smtp_password),
|
||
("SMTP_FROM_EMAIL", settings.smtp_from_email),
|
||
("REPORT_RECIPIENTS", settings.report_recipients or os.getenv("REPORT_RECIPIENTS")),
|
||
]
|
||
for key, val in required:
|
||
if not val:
|
||
missing.append(key)
|
||
return missing
|
||
|
||
|
||
def check_db_connection() -> tuple[bool, str | None]:
|
||
try:
|
||
with engine.connect() as conn:
|
||
# SQLAlchemy 2.0: wrap SQL in text() or use exec_driver_sql
|
||
conn.execute(text("SELECT 1"))
|
||
return True, None
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def check_smtp_login(s: AppSettings) -> tuple[bool, str | None]:
|
||
import smtplib
|
||
|
||
try:
|
||
server = smtplib.SMTP(s.smtp_host, s.smtp_port, timeout=20)
|
||
if s.smtp_use_tls:
|
||
server.starttls()
|
||
if s.smtp_user and s.smtp_password:
|
||
server.login(s.smtp_user, s.smtp_password)
|
||
# Probe NOOP and quit without sending
|
||
server.noop()
|
||
server.quit()
|
||
return True, None
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def main() -> int:
|
||
print("=== Workolik Production Validation ===")
|
||
settings = AppSettings()
|
||
|
||
# 1) Environment variables
|
||
missing = check_env(settings)
|
||
if missing:
|
||
print("❌ Missing required env vars:", ", ".join(sorted(set(missing))))
|
||
else:
|
||
print("✅ Required env vars present")
|
||
|
||
# Optional BCC
|
||
bcc = os.getenv("BCC_RECIPIENTS", "").strip()
|
||
if bcc:
|
||
print(f"✅ BCC_RECIPIENTS set: {bcc}")
|
||
else:
|
||
print("ℹ️ BCC_RECIPIENTS not set (no BCC will be added)")
|
||
|
||
# 2) Database connectivity
|
||
ok_db, err_db = check_db_connection()
|
||
if ok_db:
|
||
print("✅ Database connectivity OK")
|
||
else:
|
||
print(f"❌ Database connectivity FAILED: {err_db}")
|
||
|
||
# 3) SMTP connectivity (no email will be sent)
|
||
ok_smtp, err_smtp = check_smtp_login(settings)
|
||
if ok_smtp:
|
||
print("✅ SMTP login OK (no email sent)")
|
||
else:
|
||
print(f"❌ SMTP login FAILED: {err_smtp}")
|
||
|
||
# 4) Scheduler subject check (ensure dedupe matches)
|
||
today_str = datetime.now().date().strftime('%Y-%m-%d')
|
||
expected_subject = f"Daily Digest - {today_str}"
|
||
print(f"✅ Scheduler dedupe subject pattern: {expected_subject}")
|
||
|
||
failures = (1 if missing else 0) + (0 if ok_db else 1) + (0 if ok_smtp else 1)
|
||
print("=== Validation Complete ===")
|
||
return 0 if failures == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|
||
|
||
|