Coverage for an_website/quotes/quotes.py: 75.120%

209 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-10-04 17:54 +0000

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14""" 

15A page with wrong quotes. 

16 

17It displays funny, but wrong, quotes. 

18""" 

19 

20from __future__ import annotations 

21 

22import asyncio 

23import logging 

24import random 

25from asyncio import AbstractEventLoop, Future 

26from dataclasses import dataclass 

27from typing import Any, ClassVar, Final, Literal, TypeAlias 

28 

29import regex 

30from tornado.web import HTTPError 

31 

32from .. import EVENT_REDIS 

33from ..utils.data_parsing import parse_args 

34from ..utils.request_handler import APIRequestHandler, HTMLRequestHandler 

35from ..utils.utils import hash_ip 

36from .image import IMAGE_CONTENT_TYPES, create_image 

37from .quote_of_the_day import QuoteOfTheDayBaseHandler 

38from .utils import ( 

39 WRONG_QUOTES_CACHE, 

40 QuoteReadyCheckHandler, 

41 WrongQuote, 

42 create_wq_and_vote, 

43 get_authors, 

44 get_random_id, 

45 get_wrong_quote, 

46 get_wrong_quotes, 

47) 

48 

49LOGGER: Final = logging.getLogger(__name__) 

50 

51 

52RatingFilter: TypeAlias = Literal["w", "n", "unrated", "rated", "all", "smart"] 

53SMART_RATING_FILTERS: Final[tuple[RatingFilter, ...]] = ( 

54 *(("n",) * 1), 

55 *(("all",) * 5), 

56 *(("w",) * 5), 

57) 

58 

59 

60@dataclass(frozen=True) 

61class VoteArgument: 

62 """Voting matters.""" 

63 

64 vote: Literal[-1, 0, 1] 

65 

66 

67def parse_rating_filter(rating_filter_str: str) -> RatingFilter: 

68 """Get a rating filter.""" 

69 match rating_filter_str: 

70 case "w": 

71 return "w" 

72 case "n": 

73 return "n" 

74 case "unrated": 

75 return "unrated" 

76 case "rated": 

77 return "rated" 

78 case "all": 

79 return "all" 

80 return "smart" 

81 

82 

83def get_next_id(rating_filter: RatingFilter) -> tuple[int, int]: 

84 """Get the id of the next quote.""" 

85 if rating_filter == "smart": 

86 rating_filter = random.choice(SMART_RATING_FILTERS) # nosec: B311 

87 

88 match rating_filter: 

89 case "unrated": 

90 # get a random quote, but filter out already rated quotes 

91 # pylint: disable=while-used 

92 while (ids := get_random_id()) in WRONG_QUOTES_CACHE: 

93 if WRONG_QUOTES_CACHE[ids].id == -1: 

94 # check for wrong quotes that are unrated but in the cache 

95 # they don't have a real wrong_quotes_id 

96 return ids 

97 return ids 

98 case "all": 

99 return get_random_id() 

100 case "w": 

101 wrong_quotes = get_wrong_quotes(lambda wq: wq.rating > 0) 

102 case "n": 

103 wrong_quotes = get_wrong_quotes(lambda wq: wq.rating < 0) 

104 case "rated": 

105 wrong_quotes = get_wrong_quotes(lambda wq: wq.id not in {-1, None}) 

106 case _: 

107 LOGGER.error("Invalid rating filter %s", rating_filter) 

108 return get_random_id() 

109 

110 if not wrong_quotes: 

111 # invalid rating filter or no wrong quotes with that filter 

112 return get_random_id() 

113 

114 return random.choice(wrong_quotes).get_id() # nosec: B311 

115 

116 

117class QuoteBaseHandler(QuoteReadyCheckHandler): 

118 """The base request handler for the quotes package.""" 

119 

120 RATELIMIT_GET_LIMIT = 20 

121 RATELIMIT_GET_COUNT_PER_PERIOD = 20 

122 RATELIMIT_GET_PERIOD = 10 

123 

124 FUTURES: set[Future[Any]] = set() 

125 

126 loop: AbstractEventLoop 

127 next_id: tuple[int, int] 

128 rating_filter: RatingFilter 

129 

130 def future_callback(self, future: Future[WrongQuote]) -> None: 

131 """Discard the future and log the exception if one occured.""" 

132 self.FUTURES.discard(future) 

133 if exc := future.exception(): 

134 LOGGER.error( 

135 "Failed to pre-fetch quote %d-%d (%s)", 

136 *self.next_id, 

137 exc, 

138 exc_info=(type(exc), exc, exc.__traceback__), 

139 ) 

140 else: 

141 LOGGER.debug("Pre-fetched quote %d-%d", *self.next_id) 

142 

143 def get_next_url(self) -> str: 

144 """Get the URL of the next quote.""" 

145 return self.fix_url( 

146 f"/zitate/{self.next_id[0]}-{self.next_id[1]}", 

147 query_args={ 

148 "r": ( 

149 None 

150 if self.rating_filter == "smart" 

151 else self.rating_filter 

152 ), 

153 "show-rating": self.get_show_rating() or None, 

154 }, 

155 ) 

156 

157 def get_show_rating(self) -> bool: 

158 """Return whether the user wants to see the rating.""" 

159 return self.get_bool_argument("show-rating", False) 

160 

161 def on_finish(self) -> None: 

162 """ 

163 Pre-fetch the data for the next quote. 

164 

165 This is done to show the users less out-of-date data. 

166 """ 

167 if len(self.FUTURES) > 1 or ( 

168 self.content_type 

169 and self.content_type.startswith("image/") 

170 or self.content_type 

171 in {"application/pdf", "application/vnd.ms-excel"} 

172 ): 

173 return # don't spam and don't do this for images 

174 

175 user_agent = self.request.headers.get("User-Agent") 

176 if not user_agent or "bot" in user_agent.lower(): 

177 return # don't do this for bots 

178 

179 if hasattr(self, "loop"): 

180 task = self.loop.create_task( 

181 get_wrong_quote(*self.next_id, use_cache=False) 

182 ) 

183 self.FUTURES.add(task) 

184 task.add_done_callback(self.future_callback) 

185 

186 async def prepare(self) -> None: 

187 """Set the id of the next wrong_quote to show.""" 

188 await super().prepare() 

189 self.loop = asyncio.get_running_loop() 

190 self.rating_filter = parse_rating_filter(self.get_argument("r", "")) 

191 self.next_id = get_next_id(self.rating_filter) 

192 

193 

194class QuoteMainPage(QuoteBaseHandler, QuoteOfTheDayBaseHandler): 

195 """The main quote page that explains everything and links to stuff.""" 

196 

197 async def get(self, *, head: bool = False) -> None: 

198 """Render the main quote page, with a few links.""" 

199 # pylint: disable=unused-argument 

200 quote = self.get_argument("quote", "") 

201 author = self.get_argument("author", "") 

202 if (quote or author) and regex.fullmatch(r"^[0-9]+$", quote + author): 

203 self.redirect(self.fix_url(f"/zitate/{quote}-{author}")) 

204 return 

205 

206 await self.render( 

207 "pages/quotes/main_page.html", 

208 funny_quote_url=self.id_to_url( 

209 *( 

210 get_wrong_quotes(lambda wq: wq.rating > 0, shuffle=True)[ 

211 0 

212 ].get_id() 

213 ), 

214 rating_param="w", 

215 ), 

216 random_quote_url=self.id_to_url(*self.next_id), 

217 quote_of_the_day=self.redis and await self.get_quote_of_today(), # type: ignore[truthy-bool] 

218 one_stone_url=self.get_author_url("Albert Einstein"), 

219 kangaroo_url=self.get_author_url("Das Känguru"), 

220 muk_url=self.get_author_url("Marc-Uwe Kling"), 

221 ) 

222 

223 def get_author_url(self, author_name: str) -> None | str: 

224 """Get the info URL of an author.""" 

225 authors = get_authors(lambda _a: _a.name.lower() == author_name.lower()) 

226 if not authors: 

227 return None 

228 return self.fix_url(f"/zitate/info/a/{authors[0].id}") 

229 

230 def id_to_url( 

231 self, quote_id: int, author_id: int, rating_param: None | str = None 

232 ) -> str: 

233 """Get the URL of a quote.""" 

234 return self.fix_url( 

235 f"/zitate/{quote_id}-{author_id}", 

236 query_args={"r": rating_param}, 

237 ) 

238 

239 

240def wrong_quote_to_json( 

241 wq_: WrongQuote, vote: int, next_: tuple[int, int], rating: str, full: bool 

242) -> dict[str, Any]: # TODO: improve lazy return type typing 

243 """Convert a wrong quote to a dict.""" 

244 next_q, next_a = next_ 

245 if full: 

246 return { 

247 "wrong_quote": wq_.to_json(), 

248 "next": f"{next_q}-{next_a}", 

249 "vote": vote, 

250 } 

251 return { 

252 "id": wq_.get_id_as_str(), 

253 "quote": str(wq_.quote), 

254 "author": str(wq_.author), 

255 "real_author": str(wq_.quote.author), 

256 "real_author_id": wq_.quote.author_id, 

257 "rating": rating, 

258 "vote": vote, 

259 "next": f"{next_q}-{next_a}", 

260 } 

261 

262 

263class QuoteById(QuoteBaseHandler): 

264 """The page with a specified quote that then gets rendered.""" 

265 

266 RATELIMIT_POST_LIMIT: ClassVar[int] = 10 

267 RATELIMIT_POST_COUNT_PER_PERIOD: ClassVar[int] = 5 

268 RATELIMIT_POST_PERIOD: ClassVar[int] = 10 

269 

270 POSSIBLE_CONTENT_TYPES: ClassVar[tuple[str, ...]] = ( 

271 *HTMLRequestHandler.POSSIBLE_CONTENT_TYPES, 

272 *APIRequestHandler.POSSIBLE_CONTENT_TYPES, 

273 *IMAGE_CONTENT_TYPES, 

274 ) 

275 

276 LONG_PATH: ClassVar[str] = "/zitate/%d-%d" 

277 

278 async def get( 

279 self, quote_id: str, author_id: None | str = None, *, head: bool = False 

280 ) -> None: 

281 """Handle GET requests to this page and render the quote.""" 

282 int_quote_id = int(quote_id) 

283 if author_id is None: 

284 wqs = get_wrong_quotes(lambda wq: wq.id == int_quote_id) 

285 if not wqs: 

286 raise HTTPError(404, f"No wrong quote with id {quote_id}") 

287 return self.redirect(self.fix_url(self.LONG_PATH % wqs[0].get_id())) 

288 

289 if head: 

290 return 

291 

292 if ( 

293 self.content_type 

294 and self.content_type.startswith("image/") 

295 or self.content_type 

296 in {"application/pdf", "application/vnd.ms-excel"} 

297 ): 

298 wrong_quote = await get_wrong_quote(int_quote_id, int(author_id)) 

299 return await self.finish( 

300 await asyncio.to_thread( 

301 create_image, 

302 wrong_quote.quote.quote, 

303 wrong_quote.author.name, 

304 wrong_quote.rating, 

305 f"{self.request.host_name}/z/{wrong_quote.get_id_as_str(True)}", 

306 ( 

307 self.content_type.removeprefix("image/").removeprefix( 

308 "x-" 

309 ) 

310 if self.content_type.startswith("image/") 

311 else { 

312 "application/pdf": "pdf", 

313 "application/vnd.ms-excel": "xlsx", 

314 }[self.content_type] 

315 ), 

316 wq_id=wrong_quote.get_id_as_str(), 

317 ) 

318 ) 

319 

320 await self.render_quote(int_quote_id, int(author_id)) 

321 

322 async def get_old_vote( 

323 self, quote_id: int, author_id: int 

324 ) -> Literal[-1, 0, 1]: 

325 """Get the old vote from the saved vote.""" 

326 old_vote = await self.get_saved_vote(quote_id, author_id) 

327 if old_vote is None: 

328 return 0 

329 return old_vote 

330 

331 async def get_rating_str(self, wrong_quote: WrongQuote) -> str: 

332 """Get the rating str to display on the page.""" 

333 if wrong_quote.id in {None, -1}: 

334 return "---" 

335 if ( 

336 not self.get_show_rating() # don't hide the rating on wish of user 

337 and self.rating_filter == "smart" 

338 and self.request.method 

339 and self.request.method.upper() == "GET" 

340 and await self.get_saved_vote( 

341 wrong_quote.quote_id, wrong_quote.author_id 

342 ) 

343 is None 

344 ): 

345 return "???" 

346 return str(wrong_quote.rating) 

347 

348 def get_redis_votes_key(self, quote_id: int, author_id: int) -> str: 

349 """Get the key to save the votes with Redis.""" 

350 return ( 

351 f"{self.redis_prefix}:quote-votes:" 

352 f"{self.get_user_id()}:{quote_id}-{author_id}" 

353 ) 

354 

355 async def get_saved_vote( 

356 self, quote_id: int, author_id: int 

357 ) -> None | Literal[-1, 0, 1]: 

358 """ 

359 Get the vote of the current user saved with Redis. 

360 

361 Use the quote_id and author_id to query the vote. 

362 Return None if nothing is saved. 

363 """ 

364 if not EVENT_REDIS.is_set(): 

365 LOGGER.warning("No Redis connection") 

366 return 0 

367 result = await self.redis.get( 

368 self.get_redis_votes_key(quote_id, author_id) 

369 ) 

370 if result == "-1": 

371 return -1 

372 if result == "0": 

373 return 0 

374 if result == "1": 

375 return 1 

376 return None 

377 

378 @parse_args(type_=VoteArgument) 

379 async def post( 

380 self, 

381 quote_id_str: str, 

382 author_id_str: str | None = None, 

383 *, 

384 args: VoteArgument, 

385 ) -> None: 

386 """ 

387 Handle POST requests to this page and render the quote. 

388 

389 This is used to vote the quote, without changing the URL. 

390 """ 

391 quote_id = int(quote_id_str) 

392 if author_id_str is None: 

393 wqs = get_wrong_quotes(lambda wq: wq.id == quote_id) 

394 if not wqs: 

395 raise HTTPError(404, f"No wrong quote with id {quote_id}") 

396 return self.redirect(self.fix_url(self.LONG_PATH % wqs[0].get_id())) 

397 

398 author_id = int(author_id_str) 

399 

400 new_vote_str = self.get_argument("vote", None) 

401 

402 if not new_vote_str: 

403 return await self.render_quote(quote_id, author_id) 

404 

405 old_vote: Literal[-1, 0, 1] = await self.get_old_vote( 

406 quote_id, author_id 

407 ) 

408 

409 new_vote: Literal[-1, 0, 1] = args.vote 

410 vote_diff: int = new_vote - old_vote 

411 

412 if not vote_diff: # == 0 

413 return await self.render_quote(quote_id, author_id) 

414 

415 await self.update_saved_votes(quote_id, author_id, new_vote) 

416 

417 to_vote: Literal[-1, 1] = -1 if vote_diff < 0 else 1 

418 

419 contributed_by = f"an-website_{hash_ip(self.request.remote_ip, 10)}" 

420 

421 # do the voting 

422 wrong_quote = await create_wq_and_vote( 

423 to_vote, quote_id, author_id, contributed_by 

424 ) 

425 if abs(vote_diff) == 2: 

426 await wrong_quote.vote(to_vote, lazy=True) 

427 

428 await self.render_wrong_quote(wrong_quote, new_vote) 

429 

430 async def render_quote(self, quote_id: int, author_id: int) -> None: 

431 """Get and render a wrong quote based on author id and author id.""" 

432 await self.render_wrong_quote( 

433 await get_wrong_quote(quote_id, author_id), 

434 await self.get_old_vote(quote_id, author_id), 

435 ) 

436 

437 async def render_wrong_quote( 

438 self, wrong_quote: WrongQuote, vote: int 

439 ) -> None: 

440 """Render the page with the wrong_quote and this vote.""" 

441 if self.content_type in APIRequestHandler.POSSIBLE_CONTENT_TYPES: 

442 return await self.finish( 

443 wrong_quote_to_json( 

444 wrong_quote, 

445 vote, 

446 self.next_id, 

447 await self.get_rating_str(wrong_quote), 

448 self.get_bool_argument("full", False), 

449 ) 

450 ) 

451 await self.render( 

452 "pages/quotes/quotes.html", 

453 wrong_quote=wrong_quote, 

454 next_href=self.get_next_url(), 

455 next_id=f"{self.next_id[0]}-{self.next_id[1]}", 

456 description=str(wrong_quote), 

457 rating_filter=self.rating_filter, 

458 rating=await self.get_rating_str(wrong_quote), 

459 show_rating=self.get_show_rating(), 

460 vote=vote, 

461 ) 

462 

463 async def update_saved_votes( 

464 self, quote_id: int, author_id: int, vote: int 

465 ) -> None: 

466 """Save the new vote in Redis.""" 

467 if not EVENT_REDIS.is_set(): 

468 raise HTTPError(503) 

469 result = await self.redis.setex( 

470 self.get_redis_votes_key(quote_id, author_id), 

471 60 * 60 * 24 * 90, # time to live in seconds (3 months) 

472 str(vote), # value to save (the vote) 

473 ) 

474 if result: 

475 return 

476 LOGGER.warning("Could not save vote in Redis: %s", result) 

477 raise HTTPError(500, "Could not save vote") 

478 

479 

480# pylint: disable-next=too-many-ancestors 

481class QuoteAPIHandler(APIRequestHandler, QuoteById): 

482 """API request handler for the quotes page.""" 

483 

484 ALLOWED_METHODS: ClassVar[tuple[str, ...]] = ("GET", "POST") 

485 

486 POSSIBLE_CONTENT_TYPES: ClassVar[tuple[str, ...]] = ( 

487 *APIRequestHandler.POSSIBLE_CONTENT_TYPES, 

488 *IMAGE_CONTENT_TYPES, 

489 ) 

490 

491 LONG_PATH: ClassVar[str] = "/api/zitate/%d-%d" 

492 

493 async def render_wrong_quote( 

494 self, wrong_quote: WrongQuote, vote: int 

495 ) -> None: 

496 """Return the relevant data for the quotes page as JSON.""" 

497 if self.content_type == "text/plain": 

498 return await self.finish(str(wrong_quote)) 

499 return await self.finish( 

500 wrong_quote_to_json( 

501 wrong_quote, 

502 vote, 

503 self.next_id, 

504 await self.get_rating_str(wrong_quote), 

505 self.request.path.endswith("/full"), 

506 ) 

507 ) 

508 

509 

510class QuoteRedirectAPI(APIRequestHandler, QuoteBaseHandler): 

511 """Redirect to the API for a random quote.""" 

512 

513 POSSIBLE_CONTENT_TYPES = QuoteAPIHandler.POSSIBLE_CONTENT_TYPES 

514 

515 async def get( # pylint: disable=unused-argument 

516 self, suffix: str = "", *, head: bool = False 

517 ) -> None: 

518 """Redirect to a random funny quote.""" 

519 next_filter = parse_rating_filter(self.get_argument("r", "") or "w") 

520 quote_id, author_id = get_next_id(next_filter) 

521 kwargs: dict[str, str] = {"r": next_filter} 

522 if self.get_show_rating(): 

523 kwargs["show-rating"] = "sure" 

524 return self.redirect( 

525 self.fix_url( 

526 f"/api/zitate/{quote_id}-{author_id}{suffix}", 

527 query_args=kwargs, 

528 ) 

529 )