היי,
כתבתי קוד ששואב דאטה מלינק API (הלינק עצמו נמצא בדאטה בייס) ומכניס את הנתונים של הAPI לקובץ CSV.
אשמח לקבל ביקורות / שיטות יעול על הקוד.
ויש לי שאלה אולי מישהו יודע, בספיירת Stam שיצרתי קיימים יוזר וסיסמה לסרבר וכמובן ה-ip שלו.
יש דרך להצפין את הקובץ?
עריכה
אחרי ניסיון להעתיק את הקוד לתוכן (ההזחות נעלמו) צילום של הקוד (אין אפשרות לעשות קבצים בסיומת py)
שפצתי עוד קצת את הקוד:
import csv
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
import pyodbc
import requests
def run_query(conn, query):
cursor = conn.cursor()
cursor.execute(query)
return cursor.fetchall()
script_name = __file__
try:
conn = pyodbc.connect("Driver={xxx};"
"Server=xxx;"
"Database=xxx;"
"Trusted_Connection=yes;"
)
except pyodbc.InterfaceError as err:
with open('log.txt', 'a') as log:
log.write(f">>{datetime.today()}<< The connection failed, {err}, {script_name}\n")
raise
query = "exec query"
try:
link_brand = run_query(conn, query)
# link_brand = [BrandName, Link, User, Pass]
except pyodbc.ProgrammingError as err:
with open('log.txt', 'a') as log:
log.write(f">>{datetime.today()}<< Query - error in the sql code {err}, {script_name}\n")
raise
finally:
conn.close()
command = 'report'
yesterday = datetime.strftime(datetime.now() - timedelta(1), r'%m-%d-%y')
try:
with open(r'F:\myScripts\API\CX_Report.csv', 'w', newline = "") as file:
writer = csv.writer(file)
for i in link_brand:
link = f"{i[1]}?username={i[2]}&password={i[3]}&command={command}&fromdate={yesterday}&todate='{yesterday}"
response = requests.get(link)
if response.status_code != 200:
with open('log.txt', 'a') as log:
log.write(f">>{datetime.today()}<< The API of the brand {i[0]} does not work, error code {response.status_code}, {script_name}\n {link}\n")
continue
xmlContent = response.content
if xmlContent is None:
continue
soup = BeautifulSoup(xmlContent, 'xml')
for report in soup.ResultSet:
# Checking if affiliate's data is 0, if yes continue to the next one
if 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':
continue
# Titles in the DB = ["Brand", "AffiliateID", "Registrations", "FTD", "Deposits", "WithDrawals", "Comission", "Created"]
writer.writerow([i[0], # Brand
report.AffiliateID.get_text(),
report.Registrations.get_text(),
report.FTD.get_text(),
report.Deposits.get_text(),
report.WithDrawals.get_text(),
report.Comission.get_text(),
report.Day.get_text(),
])
except PermissionError:
with open('log.txt', 'a') as log:
log.write(f">>{datetime.today()}<< There is a problem with the CSV file, check if it's open and close it, {script_name}\n")
כל הכבוד על הקוד!
אני חושב שיש מקומות שבהם אפשר לנקות קצת את הקוד.
יש לי הצעה נחמדה: אני מוכן לשפר עבורך את הסקריפט בלייב הבא. נשמע דיל הוגן? 
היי ים,
תודה רבה על הפרגון, בוודאי שאשמח!
הלייב יהיה ביום שלישי? כי אני פשוט מתחתן ב1/9 וקבענו כמה עניינים לחתונה ליום שלישי… ואני לא רוצה לפספס בכדי לשמוע ביקורות או רעיונות נוספים על הקוד.
מייזל טוייף!!! 
אל דאגה, אם תראה את הלייב המוקלט ויהיו שאלות שלא כיסיתי פנה בפרטי או ע"ג הפורום ואשלים לך.
אני אהיה מחר בשיעור!
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()
מבטיח לעדכן בימים הקרובים!
עובד!! (ועוד כמה תווים)