@האדם-החושב מתנצל, אני לא כל כך מבין בזה, תוכל להסביר לי את כוונתך? שם יש טעות? להדביק את זה במקום אותם שורות בקוד ששהבאתי?
אם תוכל לומר לי בדיוק מה לשנות.
כרגע, לאחר שניסיתי לשנות את השורות שהבאת, כך נראה הקוד המלא
import requests
import csv
import io
import logging
import sys
import os
import traceback
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
# הגדרת לוגינג
log_dir = os.path.join(os.path.dirname(__file__), 'logs')
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'time_utilization.log')
logging.basicConfig(
level=logging.DEBUG, # שינוי לרמת דיבאג מלאה
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_file, encoding='utf-8', mode='w'), # מחיקת קובץ לוג קודם
logging.StreamHandler(sys.stdout)
]
)
# קבועים
TOKEN = "079607777:123456"
BASE_URL = "https://www.call2all.co.il/ym/api"
def convert_hours_to_hebrew_format(total_hours: float) -> str:
"""המרת שעות לפורמט של שעות ודקות בעברית"""
hours = int(total_hours)
minutes = int((total_hours - hours) * 60)
return f"{hours} שעות ו{minutes} דקות"
def get_hebrew_day_name(date: datetime) -> str:
"""החזרת שם יום בעברית"""
hebrew_days = {
0: "ראשון", 1: "שני", 2: "שלישי", 3: "רביעי",
4: "חמישי", 5: "שישי", 6: "שבת"
}
return hebrew_days[date.weekday()]
def get_template_entries() -> List[str]:
"""קבלת רשימת מספרי פלאפון של לומדים"""
url = f"{BASE_URL}/GetTemplateEntries"
params = {"token": TOKEN, "templateId": 153193}
try:
logging.debug(f"שולח בקשה לקבלת רשימת לומדים: {url}")
logging.debug(f"פרמטרים: {params}")
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
logging.debug(f"תגובת השרת: {response.text}")
data = response.json()
if data["responseStatus"] != "OK":
logging.error(f"שגיאה בקבלת רשימת לומדים: {data}")
return []
phones = [entry["phone"] for entry in data.get("entries", []) if not entry.get("blocked", False)]
logging.info(f"נמצאו {len(phones)} מספרי פלאפון")
return phones
except requests.exceptions.RequestException as e:
logging.error(f"שגיאת רשת בקבלת רשימת לומדים: {e}")
logging.error(traceback.format_exc())
return []
def download_ymgr_file(phone: str) -> bytes:
"""הורדת קובץ YMGR עבור מספר פלאפון"""
url = f"{BASE_URL}/RenderYMGRFile"
params = {
"token": TOKEN,
"wath": f"ivr2:/4/PrivateReports/TimeKeeperTotal.{phone}.ymgr",
"convertType": "csv",
"notLoadLang": 0
}
try:
logging.debug(f"שולח בקשה להורדת קובץ YMGR עבור {phone}: {url}")
logging.debug(f"פרמטרים: {params}")
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
logging.debug(f"גודל קובץ שהתקבל: {len(response.content)} בתים")
with open(os.path.join(log_dir, f'{phone}_ymgr.csv'), 'wb') as f:
f.write(response.content)
logging.info(f"הורדת קובץ YMGR עבור {phone} בוצעה בהצלחה")
return response.content
except requests.exceptions.RequestException as e:
logging.error(f"שגיאה בהורדת קובץ YMGR עבור {phone}: {e}")
logging.error(traceback.format_exc())
raise
def process_csv_data(csv_content: bytes, today: datetime) -> float:
"""עיבוד נתוני CSV והחזרת סך השעות עבור היום הנוכחי"""
try:
logging.debug("מתחיל לעבד קובץ CSV")
csv_text = csv_content.decode('utf-8-sig')
with open(os.path.join(log_dir, 'debug_csv.csv'), 'w', encoding='utf-8') as f:
f.write(csv_text)
reader = csv.DictReader(io.StringIO(csv_text))
total_hours = 0.0
for row in reader:
logging.debug(f"עיבוד שורה: {row}")
# בדיקת תאריך בעמודה I
entry_date_str = row['תאריך כניסה']
entry_date = datetime.strptime(entry_date_str, '%d/%m/%Y')
# בדיקה אם התאריך תואם את היום הנוכחי
if entry_date.date() == today.date():
# המרת סך השעות מעמודה M
hours_str = row['סך הזמן במשמרת']
# פיצול השעות לשעות ודקות
time_parts = hours_str.split(':')
if len(time_parts) == 3:
hours, minutes, _ = map(int, time_parts)
elif len(time_parts) == 2:
hours, minutes = map(int, time_parts)
else:
logging.warning(f"פורמט זמן לא תקין: {hours_str}")
continue
total_hours += hours + (minutes / 60)
logging.info(f"סך השעות שחושבו: {total_hours} שעות")
return total_hours
except Exception as e:
logging.error(f"שגיאה בעיבוד נתוני CSV: {e}")
logging.error(traceback.format_exc())
return 0.0
def create_tts_file(total_hours: float, today: datetime) -> str:
"""יצירת קובץ TTS עם סיכום השעות"""
try:
hebrew_day = get_hebrew_day_name(today)
hebrew_date = today.strftime("כ %d %B תשפ״ה")
hours_text = convert_hours_to_hebrew_format(total_hours)
tts_content = f"סך השעות ליום {hebrew_day} {hebrew_date}: {hours_text}"
logging.info(f"נוצר קובץ TTS: {tts_content}")
return tts_content
except Exception as e:
logging.error(f"שגיאה ביצירת קובץ TTS: {e}")
logging.error(traceback.format_exc())
return ""
def upload_tts_file(phone: str, tts_content: str):
"""העלאת קובץ TTS לשרת"""
url = f"{BASE_URL}/UploadTextFile"
params = {
"token": TOKEN,
"what": f"ivr2:5/{phone}.tts",
"contents": tts_content
}
try:
logging.debug(f"שולח בקשה להעלאת קובץ TTS עבור {phone}: {url}")
logging.debug(f"פרמטרים: {params}")
logging.debug(f"תוכן TTS: {tts_content}")
response = requests.get(url, params=params)
response.raise_for_status()
logging.debug(f"תגובת השרת בהעלאת TTS: {response.text}")
logging.info(f"העלאת קובץ TTS עבור {phone} בוצעה בהצלחה")
return response
except requests.exceptions.RequestException as e:
logging.error(f"שגיאה בהעלאת קובץ TTS עבור {phone}: {e}")
logging.error(traceback.format_exc())
raise
def update_extension(phone: str):
"""עדכון הרחבה במקרה של שגיאה"""
url = f"{BASE_URL}/UpdateExtension"
params = {
"token": TOKEN,
"path": f"ivr2:5/{phone[:6]}",
"type": "playfile",
"playfile_end_goto": "/0000/999"
}
try:
logging.debug(f"שולח בקשה לעדכון הרחבה עבור {phone}: {url}")
logging.debug(f"פרמטרים: {params}")
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
logging.debug(f"תגובת השרת בעדכון הרחבה: {response.text}")
logging.info(f"עדכון הרחבה עבור {phone} בוצע בהצלחה")
return response
except requests.exceptions.RequestException as e:
logging.error(f"שגיאה בעדכון הרחבה עבור {phone}: {e}")
logging.error(traceback.format_exc())
raise
def main():
logging.info("התחלת הרצת סקריפט ניצול זמן")
today = datetime.now()
logging.debug(f"תאריך נוכחי: {today}")
try:
phones = get_template_entries()
if not phones:
logging.warning("לא נמצאו מספרי פלאפון")
return
for phone in phones:
try:
# הורדת קובץ YMGR
ymgr_content = download_ymgr_file(phone)
# חישוב סך השעות
total_hours = process_csv_data(ymgr_content, today)
# אם אין שעות ביום הנוכחי, דלג על הלומד הזה
if total_hours == 0:
logging.info(f"אין שעות ביום הנוכחי עבור {phone}")
continue
# יצירת קובץ TTS
tts_content = create_tts_file(total_hours, today)
if not tts_content:
logging.warning(f"לא ניתן ליצור קובץ TTS עבור {phone}")
continue
# העלאת קובץ TTS
upload_tts_file(phone, tts_content)
except Exception as phone_error:
logging.error(f"שגיאה בטיפול בטלפון {phone}: {phone_error}")
logging.error(traceback.format_exc())
# במקרה של שגיאה בהעלאה, נסה לעדכן את ההרחבה ואז העלה שוב
try:
update_extension(phone)
upload_tts_file(phone, tts_content)
except Exception as retry_error:
logging.error(f"שגיאה בניסיון חוזר עבור {phone}: {retry_error}")
logging.error(traceback.format_exc())
except Exception as e:
logging.error(f"שגיאה כללית: {e}")
logging.error(traceback.format_exc())
logging.info("סיום הרצת סקריפט ניצול זמן")
if __name__ == "__main__":
main()
וזה עובד פחות מהקוד הקודם, נראה לי שכלל לא עושה כלום.
ממש ממש תודה רבה לך!!! תזכה למצוות!
