import psycopg2 as mdb
import psycopg2.extras
import smtplib
import json
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from datetime import datetime
# Disable SSL warnings (matching PHP's CURLOPT_SSL_VERIFYPEER = 0)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# --- DB connection ---
pg_config = {
'host': '37.27.12.199',
'user': 'axg_crm_new1',
'password': 'Yyv1QT2JUMLyxt71cYLG',
'dbname': 'axg_crm_new'
}
# Selenium API endpoint
SELENIUM_API_URL = 'http://135.181.199.23:5004/removee'
# Email configuration
EMAIL_USERNAME = 'AKIAYJJ2GCT4ZILVPU6C'
EMAIL_PASSWORD = 'BK/DjNVeavqxMZSBMBB/TGVbpHjiVXF7QZfF9E+NhTSh'
EMAIL_SENDER = 'removals@axghouse.com'
EMAIL_RECEIVERS = ['santhoshkasturi7@gmail.com', 'support@axghouse.com']
def check_link_status(url):
"""Check the status of a link using the Selenium API - matches PHP curl implementation"""
try:
# Match the exact PHP curl request format
payload = json.dumps({'url': url})
headers = {
'Content-Type': 'application/json'
}
response = requests.post(
SELENIUM_API_URL,
data=payload, # Use data instead of json to match curl POSTFIELDS
headers=headers,
timeout=120,
allow_redirects=True, # Equivalent to CURLOPT_FOLLOWLOCATION
verify=False # Equivalent to CURLOPT_SSL_VERIFYPEER = 0
)
# Return the response text directly (matching PHP's curl_exec return)
return response.text.strip() if response.status_code == 200 else f"API_ERROR_{response.status_code}"
except requests.exceptions.Timeout:
return "ERROR: Request timeout"
except requests.exceptions.RequestException as e:
return f"ERROR: {str(e)}"
except Exception as e:
return f"ERROR: {str(e)}"
def get_expected_codes(good_link):
"""Get expected status codes based on link type"""
if good_link == 1:
return ['200']
else:
return ['404', '403']
def verify_link(link_data):
"""Verify a single link and return result"""
url = link_data['link']
# Handle NULL/None values - default to 0 (Bad Link)
good_link = link_data.get('good_link')
if good_link is None:
good_link = 0
expected_codes = get_expected_codes(good_link)
actual_code = check_link_status(url)
is_match = actual_code in expected_codes
link_type = "Good Link" if good_link == 1 else "Bad Link"
status = "ā PASS" if is_match else "ā FAIL"
return {
'id': link_data['id'],
'url': url,
'link_type': link_type,
'expected_codes': expected_codes,
'actual_code': actual_code,
'status': status,
'is_match': is_match
}
# --- Connect to database ---
try:
con = mdb.connect(**pg_config)
con.autocommit = True
cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)
# --- Query all test links ---
query = """
SELECT
id,
link,
good_link
FROM test_links
ORDER BY id ASC;
"""
cur.execute(query)
rows = cur.fetchall()
if len(rows) == 0:
print("No test links found in database.")
con.close()
exit(0)
# --- Verify all links ---
results = []
for row in rows:
result = verify_link(row)
results.append(result)
print(f"Checked: {result['url']} - {result['status']} (Expected: {result['expected_codes']}, Got: {result['actual_code']})")
# --- Generate report statistics ---
total_links = len(results)
passed_links = sum(1 for r in results if r['is_match'])
failed_links = total_links - passed_links
good_links_count = sum(1 for r in results if r['link_type'] == 'Good Link')
bad_links_count = total_links - good_links_count
# --- Build HTML table with results ---
table_rows = ""
for result in results:
status_color = "#28a745" if result['is_match'] else "#dc3545"
table_rows += f"""
| {result['id']} |
{result['url'][:80] + '...' if len(result['url']) > 80 else result['url']} |
{result['link_type']} |
{', '.join(result['expected_codes'])} |
{result['actual_code']} |
{result['status']} |
"""
# --- Build HTML email ---
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
html = f"""
š Daily Test Links Verification Report
Total Links
{total_links}
Good Links
{good_links_count}
Bad Links
{bad_links_count}
| ID |
Link URL |
Link Type |
Expected Codes |
Actual Code |
Status |
{table_rows}
"""
# --- Send email ---
msg = MIMEMultipart("alternative")
msg["Subject"] = f"Daily Test Links Report: {passed_links}/{total_links} Passed ({failed_links} Failed)"
msg["From"] = f"Axghouse Antipiracy <{EMAIL_SENDER}>"
msg["To"] = ", ".join(EMAIL_RECEIVERS)
msg.attach(MIMEText(html, "html"))
try:
server = smtplib.SMTP('email-smtp.eu-west-1.amazonaws.com', 587)
server.starttls()
server.login(EMAIL_USERNAME, EMAIL_PASSWORD)
server.sendmail(EMAIL_SENDER, EMAIL_RECEIVERS, msg.as_string())
server.quit()
print(f"\nā
Report email sent successfully!")
print(f" Total Links: {total_links} | Passed: {passed_links} | Failed: {failed_links}")
except Exception as e:
print(f"\nā Error sending email: {str(e)}")
con.close()
except Exception as e:
print(f"ā Error: {str(e)}")
exit(1)