-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtelegram_bot_robert.py
More file actions
2510 lines (2067 loc) Β· 106 KB
/
Copy pathtelegram_bot_robert.py
File metadata and controls
2510 lines (2067 loc) Β· 106 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python
import json
import logging
import os
import re
from datetime import datetime, timedelta
import pytz
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# Path to the token file
TOKEN_FILE_PATH = "/home/users/ilo/bin/telegram_bot_robert/token"
# Enable logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Read the token from file
def read_token():
try:
with open(TOKEN_FILE_PATH, 'r') as token_file:
return token_file.read().strip()
except Exception as e:
logger.error(f"Error reading token file: {e}")
raise RuntimeError(f"Could not read token from {TOKEN_FILE_PATH}. Please check the file exists and has correct permissions.")
# Get the token
TOKEN = read_token()
# Database file
DB_FILE = "accountability_db.json"
# Default database structure
DEFAULT_DB = {
"users": {},
"weekly_logs": {},
"group_chats": {},
"edited_logs": {}, # Track edited logs
"user_settings": {}, # User preferences (reminders, goals, etc.)
"achievements": {}, # User achievements and badges
"templates": {}, # User activity templates
"activity_definitions": {}, # User-defined activity meanings
"challenges": {}, # Weekly/monthly challenges
"backups": {} # Backup metadata
}
# Timezone (change to your preferred timezone)
TIMEZONE = pytz.timezone('Europe/Helsinki')
# Load database
def load_database():
if os.path.exists(DB_FILE):
try:
with open(DB_FILE, 'r') as f:
db = json.load(f)
# Ensure the edited_logs field exists
if "edited_logs" not in db:
db["edited_logs"] = {}
return db
except json.JSONDecodeError:
logger.error("Error decoding database file")
return DEFAULT_DB.copy()
return DEFAULT_DB.copy()
# Save database
def save_database(db):
"""Save database with error handling and backup on failure."""
try:
# Create backup file first
temp_file = f"{DB_FILE}.tmp"
with open(temp_file, 'w') as f:
json.dump(db, f, indent=2)
# Atomic move to replace original file
os.rename(temp_file, DB_FILE)
logger.debug("Database saved successfully")
return True
except Exception as e:
logger.error(f"Error saving database: {e}")
# Clean up temp file if it exists
try:
if os.path.exists(temp_file):
os.remove(temp_file)
except:
pass
return False
# Get current week key (YYYY-WXX format)
def get_week_key():
now = datetime.now(TIMEZONE)
return f"{now.year}-W{now.strftime('%V')}"
# Get current day key (YYYY-M-D format)
def get_day_key():
now = datetime.now(TIMEZONE)
return f"{now.year}-{now.month}-{now.day}"
# Initialize user if not exists
def init_user(db, user_id, username):
week_key = get_week_key()
user_id_str = str(user_id)
if user_id_str not in db["users"]:
db["users"][user_id_str] = {
"username": username,
"joined_date": datetime.now(TIMEZONE).isoformat(),
"activity_totals": {},
"reminders_enabled": True, # Default to enabled
"current_streak": 0, # Current consecutive days logged
"longest_streak": 0, # Longest streak ever
"last_log_date": None, # Last date user logged (for streak calculation)
"total_logs": 0, # Total number of days logged
"achievements": [], # List of earned achievements
"goals": {}, # Weekly goals per activity
"activity_definitions": {} # User-defined activity meanings
}
if week_key not in db["weekly_logs"]:
db["weekly_logs"][week_key] = {}
if user_id_str not in db["weekly_logs"][week_key]:
db["weekly_logs"][week_key][user_id_str] = {
"logs": {},
"missed_days": []
}
# Validate activity format (1-2 letters followed by number)
def validate_activity_format(activity):
return bool(re.match(r'^[A-Za-z]{1,2}\d+$', activity))
# Parse activities from log command
def parse_activities(text):
"""Parse activities with improved validation and error handling."""
activities = {}
if not text:
return activities
parts = text.strip().split()
for part in parts:
if not validate_activity_format(part):
logger.debug(f"Skipping invalid activity format: {part}")
continue
# Extract activity letters (1-2 characters) and number
match = re.match(r'^([A-Za-z]{1,2})(\d+)$', part)
if not match:
continue
activity_letters = match.group(1).upper()
try:
value = int(match.group(2))
if value <= 0:
logger.debug(f"Skipping non-positive value: {part}")
continue
if value > 10000: # Reasonable upper limit
logger.warning(f"Large activity value detected: {part}")
# If activity already exists, sum the values
if activity_letters in activities:
activities[activity_letters] += value
logger.debug(f"Summed duplicate activity {activity_letters}: {activities[activity_letters]}")
else:
activities[activity_letters] = value
except ValueError as e:
logger.debug(f"Error parsing activity value in '{part}': {e}")
continue
return activities
# Helper functions for streak calculation
def update_user_streak(db, user_id):
"""Update user's streak based on their logging pattern."""
user_id_str = str(user_id)
today = datetime.now(TIMEZONE).date()
today_key = f"{today.year}-{today.month}-{today.day}"
user = db["users"][user_id_str]
last_log_date = user.get("last_log_date")
# Convert last log date string to date object
if last_log_date:
last_date = datetime.strptime(last_log_date, "%Y-%m-%d").date()
else:
last_date = None
# If this is the user's first log ever
if not last_date:
user["current_streak"] = 1
user["longest_streak"] = max(user.get("longest_streak", 0), 1)
user["last_log_date"] = today_key
user["total_logs"] = user.get("total_logs", 0) + 1
return
# If logging for the same day, don't change streak
if last_date == today:
return
# Calculate weekdays between last log and today
current = last_date + timedelta(days=1)
missed_weekdays = 0
while current < today:
if is_weekday(current):
missed_weekdays += 1
current += timedelta(days=1)
# If missed any weekdays, reset streak
if missed_weekdays > 0:
user["current_streak"] = 1
else:
# If today is a weekday, increment streak
if is_weekday(today):
user["current_streak"] = user.get("current_streak", 0) + 1
else:
# Weekend logs don't break or extend streaks, but they still count as a log for the day
pass
# Update longest streak if current is longer
user["longest_streak"] = max(user.get("longest_streak", 0), user["current_streak"])
user["last_log_date"] = today_key
user["total_logs"] = user.get("total_logs", 0) + 1
def check_achievements(db, user_id):
"""Check and award achievements to user."""
user_id_str = str(user_id)
user = db["users"][user_id_str]
achievements = user.get("achievements", [])
new_achievements = []
# Streak achievements
streak = user.get("current_streak", 0)
if streak >= 7 and "streak_7" not in achievements:
achievements.append("streak_7")
new_achievements.append("π₯ 7-Day Streak Master!")
if streak >= 14 and "streak_14" not in achievements:
achievements.append("streak_14")
new_achievements.append("π 2-Week Consistency Champion!")
if streak >= 30 and "streak_30" not in achievements:
achievements.append("streak_30")
new_achievements.append("π 30-Day Streak Legend!")
# Total activity achievements
total_units = sum(user.get("activity_totals", {}).values())
if total_units >= 100 and "total_100" not in achievements:
achievements.append("total_100")
new_achievements.append("π― Century Club!")
if total_units >= 500 and "total_500" not in achievements:
achievements.append("total_500")
new_achievements.append("β 500 Units Superstar!")
if total_units >= 1000 and "total_1000" not in achievements:
achievements.append("total_1000")
new_achievements.append("π 1000 Units Hall of Fame!")
# Early bird achievement (logging before 9 AM)
now = datetime.now(TIMEZONE)
if now.hour < 9 and "early_bird" not in achievements:
achievements.append("early_bird")
new_achievements.append("π
Early Bird!")
user["achievements"] = achievements
return new_achievements
def get_quick_stats(db, user_id):
"""Get quick stats for today and this week."""
user_id_str = str(user_id)
week_key = get_week_key()
day_key = get_day_key()
# Today's stats
today_activities = 0
today_units = 0
if (week_key in db["weekly_logs"] and
user_id_str in db["weekly_logs"][week_key] and
day_key in db["weekly_logs"][week_key][user_id_str]["logs"]):
daily_log = db["weekly_logs"][week_key][user_id_str]["logs"][day_key]
today_activities = len(daily_log)
today_units = sum(daily_log.values())
# Week stats
week_activities = 0
week_units = 0
week_days_logged = 0
if (week_key in db["weekly_logs"] and
user_id_str in db["weekly_logs"][week_key]):
week_logs = db["weekly_logs"][week_key][user_id_str]["logs"]
week_days_logged = len([day for day in week_logs.keys()
if is_weekday(datetime.strptime(day, "%Y-%m-%d"))])
for daily_log in week_logs.values():
week_activities += len(daily_log)
week_units += sum(daily_log.values())
return {
"today_activities": today_activities,
"today_units": today_units,
"week_activities": week_activities,
"week_units": week_units,
"week_days_logged": week_days_logged
}
# Log activities for a user
def log_activities(db, user_id, username, activities, message_id=None, chat_id=None):
"""Log activities with comprehensive error handling."""
try:
# Allow empty activities (this represents an "empty day" log)
init_user(db, user_id, username)
week_key = get_week_key()
day_key = get_day_key()
user_id_str = str(user_id)
# Get old activities for this day (if any) before updating
old_activities = db["weekly_logs"][week_key][user_id_str]["logs"].get(day_key, {})
# Check if this is a new log (not an edit)
is_new_log = day_key not in db["weekly_logs"][week_key][user_id_str]["logs"]
# Store the logs for the day
db["weekly_logs"][week_key][user_id_str]["logs"][day_key] = activities
# Track the edited message if it's provided
if message_id and chat_id:
if user_id_str not in db["edited_logs"]:
db["edited_logs"][user_id_str] = {}
msg_key = f"{chat_id}:{message_id}"
db["edited_logs"][user_id_str][msg_key] = {
"week_key": week_key,
"day_key": day_key,
"activities": activities,
"timestamp": datetime.now(TIMEZONE).isoformat()
}
# Update user totals - remove old values first
for activity, value in old_activities.items():
if activity in db["users"][user_id_str]["activity_totals"]:
db["users"][user_id_str]["activity_totals"][activity] -= value
# Prevent negative totals
if db["users"][user_id_str]["activity_totals"][activity] < 0:
logger.warning(f"Negative total for user {user_id}, activity {activity}. Resetting to 0.")
db["users"][user_id_str]["activity_totals"][activity] = 0
# Add new values
for activity, value in activities.items():
if activity not in db["users"][user_id_str]["activity_totals"]:
db["users"][user_id_str]["activity_totals"][activity] = 0
db["users"][user_id_str]["activity_totals"][activity] += value
# Update streak only for new logs (not edits)
if is_new_log:
update_user_streak(db, user_id)
# Check for new achievements (including monthly milestones)
new_achievements = check_all_achievements(db, user_id)
# Save database and return success status along with achievements
success = save_database(db)
if success:
logger.info(f"Successfully logged activities for user {username} ({user_id}): {activities}")
else:
logger.error(f"Failed to save database after logging activities for user {user_id}")
return success, new_achievements
except Exception as e:
logger.error(f"Error logging activities for user {user_id}: {e}")
return False
# Get weekly summary for a user
def get_user_weekly_summary(db, user_id):
week_key = get_week_key()
user_id_str = str(user_id)
if (week_key not in db["weekly_logs"] or
user_id_str not in db["weekly_logs"][week_key]):
return "No logs found for this week."
user_logs = db["weekly_logs"][week_key][user_id_str]["logs"]
summary = {}
# Calculate totals and find max values for each activity
for day_key, daily_log in user_logs.items():
for activity, value in daily_log.items():
if activity not in summary:
summary[activity] = {"total": 0, "max": 0, "count": 0}
summary[activity]["total"] += value
summary[activity]["count"] += 1
summary[activity]["max"] = max(summary[activity]["max"], value)
# Generate summary text
summary_text = ""
for activity, stats in summary.items():
frequency = f"{stats['count']} day{'s' if stats['count'] != 1 else ''}"
summary_text += f"{activity}: logged {frequency}, highest {stats['max']}, total {stats['total']}\n"
return summary_text if summary_text else "No activities logged this week."
# Helper function to check if a date is a weekday
def is_weekday(date):
return date.weekday() < 5 # Monday=0, Friday=4
# Update missed days in the database (weekdays only)
def update_missed_days(db):
week_key = get_week_key()
if week_key not in db["weekly_logs"]:
return
today = datetime.now(TIMEZONE)
day_of_week = today.weekday() # 0 (Monday) to 6 (Sunday)
# Don't run on Sunday (when we'll send the weekly report)
if day_of_week == 6:
return
# Check only weekdays until yesterday
yesterday = today - timedelta(days=1)
# Start from the beginning of the week (Monday)
days_since_monday = day_of_week
if day_of_week == 0: # If today is Monday, start from today
days_since_monday = 0
start_of_week = today - timedelta(days=days_since_monday)
# Loop through each weekday from Monday to yesterday
current_day = start_of_week
while current_day <= yesterday:
# Only check weekdays
if is_weekday(current_day):
check_day_key = f"{current_day.year}-{current_day.month}-{current_day.day}"
# For each user, check if they logged on this day
for user_id in db["users"]:
if (user_id in db["weekly_logs"][week_key] and
check_day_key not in db["weekly_logs"][week_key][user_id]["logs"] and
check_day_key not in db["weekly_logs"][week_key][user_id]["missed_days"]):
db["weekly_logs"][week_key][user_id]["missed_days"].append(check_day_key)
current_day += timedelta(days=1)
save_database(db)
# Send private message to user
async def send_private_message(context, user_id, text):
try:
await context.bot.send_message(chat_id=user_id, text=text)
return True
except Exception as e:
logger.error(f"Error sending private message to user {user_id}: {e}")
return False
# Command handlers
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"Start command called by user {update.effective_user.id} ({update.effective_user.username})")
await update.message.reply_text(
"Welcome to the Accountability Challenge Bot! π\n\n"
"Commands:\n"
"/log [activities] - Log your daily activities (e.g., /log M20 S30)\n"
"/status - View your current week's summary\n"
"/help - Show this help message\n\n"
"I'll remind you to log your activities and share weekly summaries!"
)
logger.info("Start command completed successfully")
except Exception as e:
logger.error(f"Error in start command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error processing your request.")
except:
pass
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"Help command called by user {update.effective_user.id} ({update.effective_user.username})")
# Check if user wants specific command help
if context.args and len(context.args) > 0:
command = context.args[0].lower()
await show_command_help(update, command)
return
# Comprehensive help listing all commands
help_text = (
"π **Accountability Challenge Bot - Complete Command Guide** π\n\n"
"**π― CORE COMMANDS**\n"
"`/log [activities]` - Log daily activities\n"
" β’ `/log M20 S30` - 20 min meditation, 30 min sport\n"
" β’ `/log KK40 P100` - 40 kickboxing, 100 pushups\n"
" β’ `/log` - Empty day (still counts for attendance!)\n\n"
"`/status` - View current week progress\n"
" β’ Shows activities, totals, missed days\n\n"
"`/help [command]` - Show this help or specific command help\n"
" β’ `/help log` - Detailed help for logging\n"
" β’ `/help goals` - Help with goals system\n\n"
"**π ANALYTICS & PROGRESS**\n"
"`/history [weeks]` - View past activity history\n"
" β’ `/history` - Last 4 weeks\n"
" β’ `/history 8` - Last 8 weeks\n\n"
"`/analytics` - Personal insights and trends\n"
" β’ Weekly trends, best days, activity patterns\n\n"
"`/level` - Check your current level and progress\n"
" β’ Shows level score, achievements, next level\n\n"
"`/calendar [month] [year]` - Visual monthly calendar\n"
" β’ `/calendar` - Current month\n"
" β’ `/calendar 12 2024` - December 2024\n\n"
"**π― GOALS & CUSTOMIZATION**\n"
"`/goals [set/remove] [activity] [target]` - Manage weekly goals\n"
" β’ `/goals` - View current goals\n"
" β’ `/goals set M 100` - Set 100 units of M per week\n"
" β’ `/goals remove M` - Remove goal for M\n\n"
"`/define [code] [description]` - Define activity meanings\n"
" β’ `/define` - View all definitions\n"
" β’ `/define M Meditation and mindfulness` - Define M\n\n"
"`/reminder [on/off]` - Toggle daily reminders\n"
" β’ Sends private reminders at 21:00 if not logged\n\n"
"**π§ POWER FEATURES**\n"
"`/edit [day] [activities]` - Edit past activities\n"
" β’ `/edit yesterday M30 S20` - Edit yesterday\n"
" β’ `/edit monday P50` - Edit last Monday\n"
" β’ `/edit 3 KK40` - Edit 3 days ago\n\n"
"`/template [save/use/list/delete] [name] [activities]`\n"
" β’ `/template save morning M20 S30` - Save template\n"
" β’ `/template use morning` - Use saved template\n"
" β’ `/template list` - Show all templates\n\n"
"`/export` - Export your personal data\n"
" β’ Summary of all your stats and activity\n\n"
"**π‘ MOTIVATION & EXTRAS**\n"
"`/quote` - Get daily motivation quote\n\n"
"**π
IMPORTANT RULES**\n"
"β’ Only weekdays (Mon-Fri) count for the challenge\n"
"β’ Use 1-2 letters + numbers (M20, KK40, etc.)\n"
"β’ You can edit previous logs by editing your message\n"
"β’ Empty logs (`/log`) count as participation\n\n"
"**β° AUTOMATED SCHEDULE**\n"
"β’ Sunday 6PM: Weekly celebration with stats\n"
"β’ Monday 8AM: New week motivation\n"
"β’ Weekdays 9PM: Optional daily reminders\n\n"
"π‘ **Tip:** Type `/help [command]` for detailed help on any command!"
)
await update.message.reply_text(help_text)
logger.info("Help command completed successfully")
except Exception as e:
logger.error(f"Error in help command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error processing your help request.")
except:
pass
async def show_command_help(update: Update, command: str):
"""Show detailed help for specific commands."""
help_texts = {
"log": (
"π **Detailed Help: /log Command** π\n\n"
"**Purpose:** Log your daily activities\n\n"
"**Format:** `/log [ActivityCode][Number] [ActivityCode][Number]...`\n\n"
"**Examples:**\n"
"β’ `/log M20` - 20 minutes of meditation\n"
"β’ `/log M20 S30 P50` - Multiple activities\n"
"β’ `/log KK40 MM15` - Double letter codes\n"
"β’ `/log` - Empty day (counts for attendance)\n\n"
"**Activity Code Rules:**\n"
"β’ 1-2 letters followed by a number\n"
"β’ Examples: M, S, P, KK, MM, BB\n"
"β’ Numbers can be minutes, reps, whatever you prefer\n\n"
"**Tips:**\n"
"β’ Only works on weekdays (Mon-Fri)\n"
"β’ You can edit logs by editing your message\n"
"β’ Define codes with `/define M Meditation`\n"
"β’ Set goals with `/goals set M 100`"
),
"goals": (
"π― **Detailed Help: /goals Command** π―\n\n"
"**Purpose:** Set and track weekly targets\n\n"
"**Commands:**\n"
"β’ `/goals` - View current goals and progress\n"
"β’ `/goals set [activity] [target]` - Set weekly goal\n"
"β’ `/goals remove [activity]` - Remove a goal\n\n"
"**Examples:**\n"
"β’ `/goals set M 100` - Aim for 100 M units per week\n"
"β’ `/goals set KK 200` - 200 KK units weekly\n"
"β’ `/goals remove M` - Remove meditation goal\n\n"
"**Features:**\n"
"β’ Shows current progress vs target\n"
"β’ Percentage completion display\n"
"β’ Color-coded status (β
achieved, π in progress, β behind)"
),
"template": (
"π **Detailed Help: /template Command** π\n\n"
"**Purpose:** Save and reuse activity combinations\n\n"
"**Commands:**\n"
"β’ `/template save [name] [activities]` - Save template\n"
"β’ `/template use [name]` - Use saved template\n"
"β’ `/template list` - Show all templates\n"
"β’ `/template delete [name]` - Delete template\n\n"
"**Examples:**\n"
"β’ `/template save morning M20 S30` - Save morning routine\n"
"β’ `/template save workout P100 KK40 S20`\n"
"β’ `/template use morning` - Log using morning template\n"
"β’ `/template delete workout` - Remove workout template\n\n"
"**Benefits:**\n"
"β’ Quick logging of regular routines\n"
"β’ Consistent activity combinations\n"
"β’ Saves typing time"
),
"edit": (
"βοΈ **Detailed Help: /edit Command** βοΈ\n\n"
"**Purpose:** Edit activities for past days\n\n"
"**Format:** `/edit [day] [activities]`\n\n"
"**Day Options:**\n"
"β’ `today` - Today (same as /log)\n"
"β’ `yesterday` - Previous day\n"
"β’ `monday`, `tuesday`, etc. - Specific weekday\n"
"β’ `1`, `2`, `3`, etc. - Days ago (max 7)\n\n"
"**Examples:**\n"
"β’ `/edit yesterday M30 S20` - Edit yesterday's log\n"
"β’ `/edit monday KK40` - Edit last Monday\n"
"β’ `/edit 3 P100` - Edit 3 days ago\n"
"β’ `/edit tuesday` - Clear Tuesday's log\n\n"
"**Limitations:**\n"
"β’ Only last 7 days\n"
"β’ Only weekdays (Mon-Fri)\n"
"β’ Completely replaces existing log for that day"
),
"analytics": (
"π **Detailed Help: /analytics Command** π\n\n"
"**Purpose:** Get personal insights and trends\n\n"
"**What You'll See:**\n"
"β’ Weekly trend analysis (trending up/down/steady)\n"
"β’ Most productive day of the week\n"
"β’ Activity combinations you often do together\n"
"β’ Personal statistics and averages\n"
"β’ Current level and progress\n\n"
"**Examples of Insights:**\n"
"β’ \"π Trending Up! 25% more units than last week!\"\n"
"β’ \"π Most Productive Day: Tuesday (avg 45 units)\"\n"
"β’ \"π Activity Combo: You often do M and S together!\"\n\n"
"**Data Period:**\n"
"β’ Analyzes your last 4 weeks of activity\n"
"β’ Updates in real-time as you log more"
)
}
if command in help_texts:
await update.message.reply_text(help_texts[command])
else:
await update.message.reply_text(
f"β No detailed help available for '{command}'.\n\n"
"Available detailed help topics:\n"
"β’ `/help log` - Logging activities\n"
"β’ `/help goals` - Goals system\n"
"β’ `/help template` - Templates\n"
"β’ `/help edit` - Editing past logs\n"
"β’ `/help analytics` - Analytics insights\n\n"
"Or just use `/help` for the complete command list."
)
# Motivation quotes system
MOTIVATION_QUOTES = [
"π Small steps every day lead to big changes every year!",
"πͺ You don't have to be perfect, you just have to be consistent.",
"π Progress, not perfection, is the goal.",
"π₯ Every day is a new opportunity to improve yourself.",
"β The only workout you regret is the one you didn't do.",
"π± Success is the sum of small efforts repeated day in and day out.",
"π Discipline is choosing between what you want now and what you want most.",
"π― You are what you do repeatedly. Excellence is a habit.",
"π Don't watch the clock; do what it does. Keep going.",
"π The journey of a thousand miles begins with a single step.",
"β‘ Your only limit is your mind.",
"π¦ Change happens when small improvements accumulate.",
"π Consistency is the key to achieving your goals.",
"π Believe in yourself and all that you are capable of.",
"π« Today's accomplishments were yesterday's impossibilities."
]
def get_motivation_quote():
"""Get a random motivation quote."""
import random
return random.choice(MOTIVATION_QUOTES)
async def quote_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
quote = get_motivation_quote()
await update.message.reply_text(f"π **Daily Motivation** π\n\n{quote}")
logger.info(f"Quote sent to user {update.effective_user.id}")
except Exception as e:
logger.error(f"Error in quote command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error getting your motivation quote.")
except:
pass
# Backup system functions
def create_backup(db):
"""Create a backup of the database."""
try:
timestamp = datetime.now(TIMEZONE).strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{timestamp}.json"
with open(backup_filename, 'w') as f:
json.dump(db, f, indent=2)
logger.info(f"Backup created: {backup_filename}")
return backup_filename
except Exception as e:
logger.error(f"Error creating backup: {e}")
return None
def auto_backup():
"""Perform automatic daily backup."""
try:
db = load_database()
backup_file = create_backup(db)
# Keep only last 7 backups
import glob
backups = sorted(glob.glob("backup_*.json"))
while len(backups) > 7:
oldest = backups.pop(0)
os.remove(oldest)
logger.info(f"Removed old backup: {oldest}")
return backup_file
except Exception as e:
logger.error(f"Error in auto backup: {e}")
return None
async def backup_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Manual backup command for admin users."""
try:
# Simple admin check - you can enhance this
user = update.effective_user
if user.username != "ilo": # Replace with your admin username
await update.message.reply_text("β This command is only available to administrators.")
return
db = load_database()
backup_file = create_backup(db)
if backup_file:
await update.message.reply_text(f"β
Backup created: {backup_file}")
else:
await update.message.reply_text("β Failed to create backup.")
logger.info(f"Manual backup requested by {user.username}")
except Exception as e:
logger.error(f"Error in backup command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error creating the backup.")
except:
pass
async def log_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"Log command called by user {update.effective_user.id} ({update.effective_user.username}) with args: {context.args}")
# Check if today is a weekday (Monday=0, Sunday=6)
today = datetime.now(TIMEZONE)
if today.weekday() >= 5: # Saturday=5, Sunday=6
await update.message.reply_text(
"π΄ It's the weekend! The accountability challenge only tracks weekdays (Monday-Friday).\n"
"Enjoy your rest days and see you on Monday! π"
)
return
# Allow empty logs (no arguments) - this counts as logging an "empty day"
activities = {}
if context.args:
activities_text = " ".join(context.args)
logger.info(f"Parsing activities: {activities_text}")
activities = parse_activities(activities_text)
logger.info(f"Parsed activities: {activities}")
else:
logger.info("Empty log command (no arguments) - logging empty day")
logger.info("Loading database...")
db = load_database()
logger.info("Database loaded successfully")
user = update.effective_user
# Log the activities with message ID for tracking edits
logger.info(f"Logging activities for user {user.id}: {activities}")
result = log_activities(
db,
user.id,
user.username or user.first_name,
activities,
update.message.message_id,
update.effective_chat.id
)
# Handle both old and new return formats
if isinstance(result, tuple):
success, new_achievements = result
else:
success = result
new_achievements = []
logger.info(f"Log activities result: {success}, achievements: {new_achievements}")
if not success:
logger.error("Failed to log activities")
await update.message.reply_text(
"β Sorry, there was an error saving your activities. Please try again."
)
return
# Get quick stats for response
quick_stats = get_quick_stats(db, user.id)
user_data = db["users"][str(user.id)]
# Generate response
if activities:
response_text = "β
Logged successfully:\n"
for activity, value in activities.items():
response_text += f"{activity}: {value} units\n"
# Add quick stats
response_text += f"\nπ **Quick Stats:**\n"
response_text += f"Today: {quick_stats['today_activities']} activities, {quick_stats['today_units']} units\n"
response_text += f"This week: {quick_stats['week_days_logged']}/5 days, {quick_stats['week_units']} total units\n"
# Add streak info
current_streak = user_data.get("current_streak", 0)
if current_streak > 1:
response_text += f"π₯ Current streak: {current_streak} days!\n"
else:
response_text = "π Empty day logged! Even rest days are part of the journey! π±\n"
response_text += f"\nπ This week: {quick_stats['week_days_logged']}/5 days logged"
# Add new achievements
if new_achievements:
response_text += "\n\nπ **NEW ACHIEVEMENTS!** π\n"
for achievement in new_achievements:
response_text += f"β’ {achievement}\n"
logger.info(f"Generated response: {response_text}")
# If in a group chat, send a brief acknowledgment and detailed response in private
if update.effective_chat.type in ["group", "supergroup"]:
logger.info("Sending group acknowledgment")
# Send brief acknowledgment in group
await update.message.reply_text("β
Activities logged! Check your private messages for details.")
# Send detailed confirmation in private
logger.info("Sending private message")
success = await send_private_message(context, user.id, response_text)
logger.info(f"Private message result: {success}")
# If private message failed, send full response in the group
if not success:
logger.warning("Private message failed, sending full response in group")
await update.message.reply_text(
f"I couldn't send you a private message. Here's your log confirmation:\n\n{response_text}"
)
else:
# If in private chat, just reply directly
logger.info("Sending direct reply in private chat")
await update.message.reply_text(response_text)
logger.info("Log command completed successfully")
except Exception as e:
logger.error(f"Error in log command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error processing your log request. Please try again.")
except:
pass
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"Status command called by user {update.effective_user.id} ({update.effective_user.username})")
logger.info("Loading database for status command...")
db = load_database()
logger.info("Database loaded successfully")
user = update.effective_user
username = user.username or user.first_name
logger.info(f"Processing status for user: {username}")
init_user(db, user.id, username)
logger.info("User initialized")
summary = get_user_weekly_summary(db, user.id)
logger.info(f"Generated summary: {summary}")
await update.message.reply_text(
f"π Weekly Summary for @{username}:\n\n{summary}"
)
logger.info("Status command completed successfully")
except Exception as e:
logger.error(f"Error in status command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error retrieving your status. Please try again.")
except:
pass
async def reminder_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"Reminder command called by user {update.effective_user.id} ({update.effective_user.username}) with args: {context.args}")
if not context.args or context.args[0].lower() not in ['on', 'off']:
await update.message.reply_text(
"Please specify whether to turn reminders on or off:\n"
"/reminder on - Enable daily reminders at 21:00\n"
"/reminder off - Disable daily reminders"
)
return
db = load_database()
user = update.effective_user
user_id_str = str(user.id)
# Initialize user if needed
init_user(db, user.id, user.username or user.first_name)
# Update reminder preference
enable_reminders = context.args[0].lower() == 'on'
db["users"][user_id_str]["reminders_enabled"] = enable_reminders
save_database(db)
if enable_reminders:
await update.message.reply_text(
"β
Daily reminders enabled! I'll send you a private message at 21:00 each weekday if you haven't logged yet."
)
else:
await update.message.reply_text(
"π Daily reminders disabled. You can re-enable them anytime with /reminder on"
)
logger.info(f"Reminder preference updated for user {user.id}: {enable_reminders}")
except Exception as e:
logger.error(f"Error in reminder command: {e}", exc_info=True)
try:
await update.message.reply_text("β Sorry, there was an error updating your reminder preference.")
except:
pass
async def history_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
logger.info(f"History command called by user {update.effective_user.id} ({update.effective_user.username}) with args: {context.args}")
db = load_database()
user = update.effective_user
user_id_str = str(user.id)
# Initialize user if needed
init_user(db, user.id, user.username or user.first_name)
# Determine how many weeks to show (default 4)
weeks_to_show = 4
if context.args and context.args[0].isdigit():
weeks_to_show = min(int(context.args[0]), 12) # Max 12 weeks
# Get current week and calculate previous weeks
current_week = get_week_key()
weeks_to_check = []
# Calculate week keys for the past N weeks
current_date = datetime.now(TIMEZONE)
for i in range(weeks_to_show):
week_date = current_date - timedelta(weeks=i)
week_key = f"{week_date.year}-W{week_date.strftime('%V')}"