Coverage for seedboxsync/core/database/models/torrent.py: 61%

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 Torrent.""" 

8 

9from collections.abc import Iterable 

10import datetime 

11from os import PathLike 

12from typing import Any, cast 

13from urllib.parse import urlparse 

14from peewee import AutoField, BooleanField, DateTimeField, IntegerField, TextField, fn 

15import tldextract 

16from seedboxsync.core import utils 

17from seedboxsync.core.database.models import SeedboxSyncModel 

18 

19 

20class Torrent(SeedboxSyncModel): 

21 """ 

22 Data Access Object (DAO) representing a torrent. 

23 

24 This model stores metadata for a torrent file, including tracker info, 

25 source provenance, file counts, payload size, privacy status, and processing timestamp. 

26 

27 Attributes: 

28 id (int): Auto-incremented primary key. 

29 name (str): Name of the torrent. 

30 announce (str): Tracker announce URL. 

31 announcer (str): Tracker announce domain name. 

32 source (str): Source or provenance tag of the torrent file. 

33 total_files (int): Total number of files contained in the torrent. 

34 total_size (int): Total size of all files in the torrent in bytes. 

35 private (bool): Flag indicating if the torrent is marked as private. 

36 sent (datetime): Timestamp indicating when the torrent was sent or created. 

37 """ 

38 

39 id = AutoField(help_text="Unique identifier of the torrent") 

40 name = TextField(help_text="Name of the torrent") 

41 announce = TextField(null=True, help_text="Tracker announce URL of the torrent") 

42 announcer = TextField(null=True, help_text="Tracker announce domain of the torrent") 

43 source = TextField(null=True, help_text="Source or provenance of the torrent file") 

44 total_files = IntegerField(null=True, help_text="Total number of files contained in the torrent") 

45 total_size = IntegerField(null=True, help_text="Total size of all files in bytes") 

46 private = BooleanField(null=False, default=False, help_text="Flag indicating if the torrent is private") 

47 sent = DateTimeField(default=datetime.datetime.now, help_text="Timestamp when the torrent was sent") 

48 

49 def save(self, force_insert: bool = False, only: Any | None = None) -> int: 

50 """Save the model instance, automatically updating the announcer domain.""" 

51 self.update_announcer() 

52 return super().save(force_insert=force_insert, only=only) 

53 

54 def set_from_file(self, torrent_file: str | PathLike[str]) -> bool: 

55 """ 

56 Populate the model from a torrent file. 

57 

58 Args: 

59 torrent_file: Path to the torrent file. 

60 

61 Returns: 

62 True if the torrent is valid and minimal informations was successfully extracted, 

63 otherwise False. 

64 """ 

65 torrent_info = utils.get_torrent_infos(torrent_file) 

66 # Torrent is not valid 

67 if not isinstance(torrent_info, dict): 

68 return False 

69 

70 # Get minimal information 

71 self.announce = torrent_info.get("announce") or None 

72 

73 # Get extended informations 

74 info = torrent_info.get("info") 

75 if not isinstance(info, dict): 

76 return True 

77 

78 self.source = info.get("source") or None 

79 self.private = info.get("private", False) 

80 

81 self.total_files = None 

82 self.total_size = None 

83 files = info.get("files") 

84 if isinstance(files, list): 

85 # Multi-file torrent. 

86 valid_files = [file_info for file_info in files if isinstance(file_info, dict) and isinstance(file_info.get("length"), int)] 

87 self.total_files = len(valid_files) 

88 self.total_size = sum(file_info["length"] for file_info in valid_files) 

89 else: 

90 # Single-file torrent. 

91 length = info.get("length") 

92 self.total_files = 1 if isinstance(length, int) else None 

93 self.total_size = length if isinstance(length, int) else None 

94 

95 return True 

96 

97 def update_announcer(self) -> None: 

98 """ 

99 Extract and populate the announcer domain from the announce URL. 

100 

101 Parses ``announce`` URL and extracts the registrable domain (without subdomains). 

102 """ 

103 self.announcer = None # reset on update 

104 if not self.announce: 

105 return 

106 

107 hostname = urlparse(self.announce).hostname 

108 if not hostname: 

109 return 

110 

111 # Use tldextract 

112 extracted = tldextract.extract(hostname) 

113 if not extracted.domain or not extracted.suffix: 

114 self.announcer = hostname 

115 return 

116 

117 # Set announcer 

118 self.announcer = f"{extracted.domain}.{extracted.suffix}" 

119 

120 # Source fallback 

121 if self.source == "" or self.source is None: 

122 self.source = self.announcer 

123 

124 @classmethod 

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

126 """ 

127 Get the total count and total size of torrents grouped by source. 

128 

129 Args: 

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

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

132 

133 Returns: 

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

135 """ 

136 # Build "where" expression 

137 conditions = [] 

138 if start_date: 

139 conditions.append(cls.sent >= start_date) 

140 if end_date: 

141 conditions.append(cls.sent <= end_date) 

142 

143 query = ( 

144 cls.select( 

145 cls.source, 

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

147 fn.SUM(cls.total_size).alias("total_size"), 

148 fn.humanize(fn.SUM(cls.total_size)).alias("human_total_size"), 

149 ) 

150 .group_by(cls.source) 

151 .order_by(fn.SUM(cls.total_size).desc()) 

152 ) 

153 

154 # if "where" expression 

155 if conditions: 

156 query = query.where(*conditions) 

157 data = query.dicts() 

158 

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