""" Database utilities for Link Checker package """ import psycopg2 from psycopg2.extras import RealDictCursor # Try relative import first, fall back to absolute try: from .exceptions import DatabaseError except ImportError: from exceptions import DatabaseError def get_connection(config): """ Get database connection Args: config (dict): Database configuration Returns: tuple: (connection, cursor) """ try: con = psycopg2.connect( host=config['host'], user=config['user'], password=config['password'], dbname=config['dbname'] ) # Set autocommit if specified if config.get('autocommit', True): con.autocommit = True cur = con.cursor(cursor_factory=RealDictCursor) return con, cur except Exception as e: raise DatabaseError(f"Failed to connect to database: {e}") def get_website_settings(cur, display_link): """ Get website settings from database Args: cur: Database cursor display_link (str): Display link Returns: dict: Website settings """ try: # Check for merged websites sql = "SELECT * FROM merge_websites" cur.execute(sql) merges = cur.fetchall() for merge in merges: website_merge = merge['site_url'] awebsite_url = '.' + website_merge if awebsite_url in display_link: display_link = website_merge break # Get website settings sql = "SELECT * FROM websites WHERE website_site_url = %s" cur.execute(sql, [display_link]) websites = cur.fetchall() if not cur.rowcount: return { 'display_link': display_link, 'website_settings': {}, 'cloudflare': False, 'average_removal_time': None, 'hosting_email': None } website_row = websites[0] website_settings = { 'check_http_code': website_row.get('check_http_code'), 'http_codes': website_row.get('http_codes'), 'check_stop_words': website_row.get('check_stop_words'), 'stopwords': website_row.get('stopwords'), 'homepage_redirect': website_row.get('homepage_redirect'), 'check_stop_words_on_page': website_row.get('check_stop_words_on_page'), 'use_selenium': website_row.get('use_selenium'), 'use_playwright': website_row.get('use_playwright') } hosting_email = website_row.get('website_hosting_email') cloudflare = False if hosting_email and 'cloudflare.com' in hosting_email: cloudflare = True return { 'display_link': display_link, 'website_settings': website_settings, 'cloudflare': cloudflare, 'average_removal_time': website_row.get('average_removal_time'), 'hosting_email': hosting_email } except Exception as e: raise DatabaseError(f"Failed to get website settings: {e}") def get_content_type_settings(cur, project_id): """ Get content type settings for a project Args: cur: Database cursor project_id (int): Project ID Returns: dict: Content type settings """ try: sql = "SELECT * FROM project_content_types WHERE project_id = %s" cur.execute(sql, [project_id]) if not cur.rowcount: return {} rows = cur.fetchall() content_type_id = rows[0]['content_type_id'] sql = "SELECT * FROM content_types WHERE content_type_id = %s" cur.execute(sql, [str(content_type_id)]) rowss = cur.fetchall() if not cur.rowcount: return {} content_type_row = rowss[0] return { 'check_http_code': content_type_row.get('check_http_code'), 'http_codes': content_type_row.get('http_codes'), 'check_stop_words': content_type_row.get('check_stop_words'), 'stop_words': content_type_row.get('stop_words'), 'homepage_redirect': content_type_row.get('homepage_redirect'), 'check_stop_words_on_page': content_type_row.get('check_stop_words_on_page'), 'use_selenium': content_type_row.get('use_selenium'), 'use_playwright': content_type_row.get('use_playwright') } except Exception as e: raise DatabaseError(f"Failed to get content type settings: {e}") def get_global_deletion_settings(cur): """ Get global deletion settings from database Args: cur: Database cursor Returns: dict: Global settings """ try: sql = "SELECT * FROM global_deletion_settings WHERE is_active = 1" cur.execute(sql) rows = cur.fetchall() global_settings = { 'http_code': [], 'stopword': [], 'redirect': [], 'skip_code': [] } for row in rows: category = row['category'] key_value = row['key_value'] cloudflare = row.get('cloudflare', 0) if category in global_settings: # For skip_code, include cloudflare flag if category == 'skip_code': global_settings[category].append({ 'code': key_value, 'cloudflare': cloudflare }) else: global_settings[category].append(key_value) return global_settings except Exception as e: raise DatabaseError(f"Failed to get global deletion settings: {e}") def get_project(cur, project_id): """ Get project information Args: cur: Database cursor project_id (int): Project ID Returns: dict: Project information """ try: sql = "SELECT * FROM projects WHERE project_id = %s" cur.execute(sql, [project_id]) rows = cur.fetchall() return rows[0] if cur.rowcount > 0 else {} except Exception as e: raise DatabaseError(f"Failed to get project: {e}") def get_project_title_score(cur): """ Get project title score Args: cur: Database cursor project_id (int): Project ID Returns: dict: Project information """ try: return 40 # Default value if not set # sql = "SELECT value FROM tse_configs WHERE component = 'FWS' and config = 'project_title_score'" # cur.execute(sql) # row = cur.fetchone() # return int(row['value']) if row else 70 except Exception as e: raise DatabaseError(f"Failed to get project: {e}") def get_user_agent(cur): """ Get random user agent Args: cur: Database cursor Returns: str: User agent string """ try: sql = 'SELECT * FROM user_agents WHERE "deviceCategory" = \'desktop\' ORDER BY RANDOM() LIMIT 1' cur.execute(sql) agents = cur.fetchall() return agents[0]['useragent'] if cur.rowcount > 0 else None except Exception as e: raise DatabaseError(f"Failed to get user agent: {e}") def get_cookies(cur, table): """ Get cookies for social media platforms Args: cur: Database cursor table (str): Cookie table name Returns: list: Cookie records """ try: sql = "SELECT * FROM " + table + " LIMIT 1" cur.execute(sql) rows = cur.fetchall() return rows except Exception as e: raise DatabaseError(f"Failed to get cookies: {e}") def get_deleted_keywords(cur): """ Get deleted keywords for content checking Args: cur: Database cursor Returns: list: Deleted keywords """ try: import json user_group_id = 7 sql = "SELECT * FROM user_groups WHERE user_group_id = %s" cur.execute(sql, [user_group_id]) rows = cur.fetchall() settings = rows[0]['settings'] keywords = json.loads(settings) deleted_keywords = keywords['keywords_detecting_remove_pirated_content'].split('\r\n') return deleted_keywords except Exception as e: raise DatabaseError(f"Failed to get deleted keywords: {e}") def is_cloudflare_website(hosting_email): """ Check if website is cloudflare based on hosting email Args: hosting_email (str): Hosting email from database Returns: bool: True if cloudflare, False otherwise """ if not hosting_email: return False return 'cloudflare.com' in hosting_email.lower()