import csv
import logging
from datetime import datetime, timedelta
from typing import List, NamedTuple, TYPE_CHECKING, TextIO
from bs4 import BeautifulSoup # type: ignore
import pyodbc # type: ignore
import requests
if TYPE_CHECKING:
import _csv
logger = logging.getLogger(__name__)
# Constants
CSV_PATH = r'F:\myScripts\API\CX_Report.csv'
LINK_BRANDS_QUERY = 'SELECT * FROM Users;'
REPORT_COMMAND = 'report'
# Configurations
class ServerDetails(NamedTuple):
driver: str
server: str
database: str
def get_db_connection(
driver: str, server: str, database: str,
) -> pyodbc.Connection:
try:
return pyodbc.connect(
f'Driver={driver};Server={server};'
f'Database={database};Trusted_Connection=yes;',
)
except pyodbc.InterfaceError:
logger.exception('The connection failed.')
raise
def run_query(query: str, server_details: ServerDetails):
conn = get_db_connection(*server_details)
cursor = conn.cursor()
cursor.execute(query)
try:
results = cursor.fetchall()
except pyodbc.ProgrammingError:
logger.exception('Query - error in the sql code.')
raise
finally:
conn.close()
return results
def get_yesterday() -> str:
return datetime.strftime(datetime.now() - timedelta(1), r'%m-%d-%y')
def safely_get_file(path: str) -> TextIO:
try:
return open(path, 'w', newline='')
except PermissionError:
logger.exception(
'There is a problem with the CSV file,'
"check if it's open and close it",
)
raise
def get_link(
link: str, username: str, password: str, command: str, yesterday: str,
) -> str:
return (
f'{link}'
f'?username={username}&password={password}'
f'&command={command}'
f"&fromdate={yesterday}&todate='{yesterday}"
)
def is_affiliate_data_zero(report) -> bool:
return (
report.Registrations.get_text() == '0'
and report.FTD.get_text() == '0'
and report.Deposits.get_text() == '0'
and report.WithDrawals.get_text() == '0'
and report.Comission.get_text() == '0'
)
def get_report_data(report) -> List[str]:
affiliate_id = report.AffiliateID.get_text()
registrations = report.Registrations.get_text()
ftd = report.FTD.get_text()
deposits = report.Deposits.get_text()
withdrawls = report.WithDrawals.get_text()
comission = report.Comission.get_text()
created = report.Day.get_text()
return [
brand_name, affiliate_id, registrations, ftd,
deposits, withdrawls, comission, created,
]
def write_brands_to_file(
writer: '_csv._writer', link: str, brand_name: str,
) -> None:
response = requests.get(link)
if not response.ok:
logger.error(
f'The API of the brand {brand_name} does not work,'
f'error code {response.status_code} for {link}',
)
return
xmlContent = response.content
if not xmlContent:
return
soup = BeautifulSoup(xmlContent, 'xml')
for report in soup.ResultSet:
if not is_affiliate_data_zero(report):
writer.writerow(get_report_data(report))
if __name__ == '__main__':
server_details = ServerDetails(driver='', server='', database='')
link_brand = run_query(LINK_BRANDS_QUERY, server_details)
file = safely_get_file(CSV_PATH)
writer = csv.writer(file)
for brand_name, link, user, password in link_brand:
link = get_link(link, user, password, REPORT_COMMAND, get_yesterday())
write_brands_to_file(writer, link, brand_name)
file.close()