Coverage for seedboxsync/core/database/models/download.py: 69%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""Peewee DAO model for Download.""" 

8 

9from collections.abc import Iterable 

10import datetime 

11from typing import Any, cast 

12from peewee import AutoField, CharField, DateTimeField, IntegerField, TextField, fn 

13from seedboxsync.core import utils 

14from seedboxsync.core.database.models import SeedboxSyncModel 

15 

16 

17class Download(SeedboxSyncModel): 

18 """ 

19 Data Access Object (DAO) representing a file download. 

20 

21 This model stores information about a downloaded file, including its path, 

22 size on the seedbox and locally, as well as timestamps indicating when 

23 the download started and finished. 

24 """ 

25 

26 id = AutoField(help_text="Unique identifier of the download") 

27 path = TextField(help_text="Path of the downloaded file") 

28 seedbox_size = IntegerField(help_text="Size of the file on the seedbox in bytes") 

29 local_size = IntegerField(default=0, help_text="Size of the downloaded file stored locally in bytes") 

30 mime_extension = CharField(max_length=255, default="", help_text="File extension detected during MIME analysis") 

31 mime_type = CharField(max_length=100, default="", help_text="MIME type detected for the file (e.g. video/mp4)") 

32 mime_confidence = CharField(max_length=65, default="", help_text="Confidence level or method used for MIME detection (e.g. extension, magic)") 

33 started = DateTimeField(default=datetime.datetime.now, help_text="Timestamp when the download started") 

34 finished = DateTimeField(default=0, help_text="Timestamp when the download finished") 

35 

36 def set_mime(self, save: bool = False) -> None: 

37 """ 

38 Detect and update the MIME attributes of the downloaded file. 

39 

40 Calls ``utils.get_mime_type_from_file`` to inspect the file path and header bytes, 

41 updating the ``mime_extension``, ``mime_type``, and ``mime_confidence`` attributes. 

42 

43 Args: 

44 save (bool, optional): If True, saves the model instance to the database 

45 immediately after updating attributes. Defaults to False. 

46 """ 

47 self.mime_type, self.mime_extension, self.mime_confidence = utils.get_mime_type_from_file(self.path) 

48 

49 if save: 

50 self.save() 

51 

52 @classmethod 

53 def is_already_download(cls, filepath: str) -> bool: 

54 """ 

55 Check if a file has already been downloaded. 

56 

57 Args: 

58 filepath (str): Absolute or relative path to the file. 

59 

60 Returns: 

61 bool: True if the file was already downloaded (i.e. has a nonzero 

62 ``finished`` timestamp), otherwise False. 

63 """ 

64 count = cls.select().where(cls.path == filepath, cls.finished > 0).count() 

65 return count != 0 

66 

67 @classmethod 

68 def get_stats_by_mime_type(cls, start_date: datetime.date | None = None, end_date: datetime.date | None = None) -> list[dict[str, Any]]: 

69 """ 

70 Get the total count and total size of downloaded files grouped by MIME type. 

71 

72 Args: 

73 start_date (datetime.date | None): Optional start date filter. 

74 end_date (datetime.date | None): Optional end date filter. 

75 

76 Returns: 

77 list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size. 

78 """ 

79 # Build "where" expression 

80 conditions = [] 

81 if start_date: 

82 conditions.append(cls.finished >= start_date) 

83 if end_date: 

84 conditions.append(cls.finished <= end_date) 

85 

86 query = ( 

87 cls.select( 

88 cls.mime_type, 

89 fn.COUNT(cls.id).alias("total"), 

90 fn.SUM(cls.local_size).alias("total_size"), 

91 fn.humanize(fn.SUM(cls.local_size)).alias("human_total_size"), 

92 ) 

93 .group_by(cls.mime_type) 

94 .order_by(fn.SUM(cls.local_size).desc()) 

95 ) 

96 

97 # if "where" expression 

98 if conditions: 

99 query = query.where(*conditions) 

100 data = query.dicts() 

101 

102 return list(cast(Iterable[dict[str, Any]], data))