Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/backups.py: 15%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5"""Project level backups.""" 

6 

7from __future__ import annotations 

8 

9import json 

10import os 

11import warnings 

12from collections import defaultdict 

13from datetime import datetime 

14from functools import partial 

15from itertools import chain 

16from operator import itemgetter 

17from pathlib import Path 

18from shutil import copyfileobj 

19from typing import TYPE_CHECKING, Any, BinaryIO, TypedDict 

20from zipfile import ZipFile 

21 

22from django.conf import settings 

23from django.core.files import File 

24from django.db import connection, transaction 

25from django.db.models.fields.files import FieldFile 

26from django.db.models.signals import pre_save 

27from django.utils import timezone 

28from django.utils.timezone import make_aware 

29from weblate_schemas import load_schema, validate_schema 

30 

31from weblate.auth.models import AutoGroup, Group, Role, User, get_anonymous 

32from weblate.checks.models import Check 

33from weblate.lang.models import Language, Plural 

34from weblate.memory.models import Memory 

35from weblate.screenshots.models import Screenshot 

36from weblate.trans.models import ( 

37 Category, 

38 Comment, 

39 Component, 

40 Label, 

41 PendingUnitChange, 

42 Project, 

43 Suggestion, 

44 Translation, 

45 Unit, 

46 Vote, 

47) 

48from weblate.utils.data import data_path 

49from weblate.utils.hash import checksum_to_hash, hash_to_checksum 

50from weblate.utils.validators import validate_filename 

51from weblate.utils.version import VERSION 

52from weblate.vcs.models import VCS_REGISTRY 

53 

54if TYPE_CHECKING: 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true

55 from collections.abc import Callable 

56 

57 from django.db.models import Model 

58 

59 from weblate.billing.models import Billing 

60 

61warnings.filterwarnings("error", module="zipfile") 

62 

63PROJECTBACKUP_PREFIX = "projectbackups" 

64 

65 

66class BackupListDict(TypedDict): 

67 name: str 

68 path: str 

69 timestamp: datetime 

70 size: int 

71 

72 

73def list_backups(project_id: Project | int | str) -> list[BackupListDict]: 

74 if isinstance(project_id, Project): 

75 project_id = project_id.pk 

76 backup_dir = data_path(PROJECTBACKUP_PREFIX) / f"{project_id}" 

77 if not backup_dir.exists(): 

78 return [] 

79 result: list[BackupListDict] = [ 

80 { 

81 "name": entry.name, 

82 "path": entry.as_posix(), 

83 "timestamp": make_aware( 

84 datetime.fromtimestamp( # noqa: DTZ006 

85 int(entry.name.split(".")[0]) 

86 ) 

87 ), 

88 "size": entry.stat().st_size, 

89 } 

90 for entry in backup_dir.glob("*.zip") 

91 ] 

92 return sorted(result, key=itemgetter("timestamp"), reverse=True) 

93 

94 

95class ProjectBackup: 

96 COMPONENTS_PREFIX = "components/" 

97 VCS_PREFIX = "vcs/" 

98 VCS_PREFIX_LEN = len(VCS_PREFIX) 

99 

100 def __init__(self, filename: str = "", *, fileio: BinaryIO | None = None) -> None: 

101 self.data: dict[str, Any] = {} 

102 self.filename = filename 

103 self.fileio = fileio 

104 self.timestamp = timezone.now() 

105 self.project: Project | None = None 

106 self.project_schema = load_schema("weblate-backup.schema.json") 

107 self.component_schema = load_schema("weblate-component.schema.json") 

108 self.languages_cache: dict[str, Language] = {} 

109 self.labels_map: dict[str, Label] = {} 

110 self.user_cache: dict[str, User] = {} 

111 self.components_cache: dict[str, Component] = {} 

112 self.categories_cache: dict[str, Category] = {} 

113 self.roles_cache: dict[str, Role] = {} 

114 

115 @staticmethod 

116 def full_slug_without_project(obj: Component | Category) -> str: 

117 """Return the full slug for a component or category without the project slug.""" 

118 parts = obj.get_url_path() 

119 return "/".join(parts[1:]) 

120 

121 @property 

122 def supports_restore(self) -> bool: 

123 return connection.features.can_return_rows_from_bulk_insert 

124 

125 def validate_data(self) -> None: 

126 validate_schema(self.data, "weblate-backup.schema.json") 

127 

128 def backup_property( 

129 self, obj: Model, field: str, extras: dict[str, Callable] | None = None 

130 ) -> str | int | dict | None: 

131 if extras and field in extras: 

132 return extras[field](obj) 

133 value = getattr(obj, field) 

134 if isinstance(value, Language): 

135 return value.code 

136 if isinstance(value, Plural): 

137 return self.backup_object( 

138 value, 

139 self.component_schema["properties"]["translations"]["items"][ 

140 "properties" 

141 ]["plural"]["required"], 

142 ) 

143 if isinstance(value, Unit): 

144 return value.checksum 

145 if isinstance(value, User): 

146 return value.username 

147 if isinstance(value, datetime): 

148 return value.isoformat() 

149 if isinstance(value, FieldFile): 

150 return os.path.basename(value.name) # type: ignore[type-var] 

151 return value 

152 

153 def backup_object( 

154 self, 

155 obj: Model, 

156 properties: list[str], 

157 extras: dict[str, Callable] | None = None, 

158 ) -> dict[str, str | int | dict | None]: 

159 return {field: self.backup_property(obj, field, extras) for field in properties} 

160 

161 def backup_m2m_flat(self, obj: Model, relation: str, field: str) -> list: 

162 """Backup a many to many relation using a unique identifying field of the related object.""" 

163 return list(getattr(obj, relation).values_list(field, flat=True)) 

164 

165 def backup_teams(self, project: Project) -> list[dict]: 

166 extras: dict[str, Callable] = {} 

167 for schema_name, relation, field in [ 

168 ("roles", "roles", "name"), 

169 ("languages", "languages", "code"), 

170 ("admins", "admins", "username"), 

171 ("members", "user_set", "username"), 

172 ("autogroups", "autogroup_set", "match"), 

173 ]: 

174 extras[schema_name] = partial( 

175 self.backup_m2m_flat, 

176 relation=relation, 

177 field=field, 

178 ) 

179 extras["components"] = lambda obj: [ 

180 self.full_slug_without_project(c) for c in obj.components.all() 

181 ] 

182 

183 return [ 

184 self.backup_object( 

185 group, 

186 self.project_schema["properties"]["teams"]["items"]["required"], 

187 extras, 

188 ) 

189 for group in project.defined_groups.all() 

190 ] 

191 

192 def backup_categories(self, obj: Project | Category) -> list[dict]: 

193 if isinstance(obj, Project): 

194 categories = obj.category_set.filter(category=None) 

195 else: 

196 categories = obj.category_set.all() 

197 return [ 

198 self.backup_object( 

199 category, 

200 self.project_schema["definitions"]["category"]["required"], 

201 {"categories": self.backup_categories}, 

202 ) 

203 for category in categories 

204 ] 

205 

206 def backup_data(self, project: Project) -> None: 

207 self.project = project 

208 self.data = { 

209 "metadata": { 

210 "version": VERSION, 

211 "server": settings.SITE_TITLE, 

212 "domain": settings.SITE_DOMAIN.rsplit(":", 1)[0], 

213 "timestamp": self.timestamp.isoformat(), 

214 }, 

215 "project": self.backup_object( 

216 project, self.project_schema["properties"]["project"]["required"] 

217 ), 

218 "labels": [ 

219 {"name": label.name, "color": label.color} 

220 for label in project.label_set.all() 

221 ], 

222 "categories": self.backup_categories(project), 

223 "teams": self.backup_teams(project), 

224 } 

225 

226 # Make sure generated backup data is correct 

227 self.validate_data() 

228 

229 def backup_dir(self, backupzip: ZipFile, directory: str, target: str) -> None: 

230 """Backup single directory to specified target in zip.""" 

231 for folder, _subfolders, filenames in os.walk(directory): 

232 for filename in filenames: 

233 path = os.path.join(folder, filename) 

234 # zipfile does not support storing symlinks, it dereferences them 

235 if os.path.islink(path): 

236 continue 

237 backupzip.write( 

238 path, os.path.join(target, os.path.relpath(path, directory)) 

239 ) 

240 

241 def backup_json(self, backupzip: ZipFile, data: dict | list, target: str) -> None: 

242 with backupzip.open(target, "w") as handle: 

243 handle.write(json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")) 

244 

245 def generate_filename(self, project: Project) -> None: 

246 # Create directory 

247 backup_dir = data_path(PROJECTBACKUP_PREFIX) / f"{project.pk}" 

248 backup_dir.mkdir(parents=True, exist_ok=True) 

249 

250 # Create README.txt 

251 backup_info = backup_dir / "README.txt" 

252 if not backup_info.exists(): 

253 with backup_info.open("w") as handle: 

254 handle.write(f"# Weblate project backups for {project.name}\n") 

255 handle.write(f"slug={project.slug}\n") 

256 handle.write(f"web={project.web}\n") 

257 handle.writelines( 

258 f"billing={billing.id}\n" for billing in project.billings 

259 ) 

260 

261 # Find unused timestamp 

262 timestamp = int(self.timestamp.timestamp()) 

263 while (filename := backup_dir / f"{timestamp}.zip").exists() or ( 

264 backup_dir / f"{timestamp}.zip.part" 

265 ).exists(): 

266 timestamp += 1 

267 

268 self.filename = filename.as_posix() 

269 

270 def backup_component(self, backupzip: ZipFile, component: Component) -> None: 

271 data: dict = { 

272 "component": self.backup_object( 

273 component, self.component_schema["properties"]["component"]["required"] 

274 ), 

275 "translations": [ 

276 self.backup_object( 

277 translation, 

278 self.component_schema["properties"]["translations"]["items"][ 

279 "required" 

280 ], 

281 ) 

282 for translation in component.translation_set.iterator() 

283 ], 

284 "units": [ 

285 self.backup_object( 

286 unit, 

287 self.component_schema["properties"]["units"]["items"]["required"], 

288 extras={ 

289 "id_hash": lambda obj: obj.checksum, 

290 "comments": lambda obj: [ 

291 self.backup_object( 

292 comment, 

293 self.component_schema["properties"]["units"]["items"][ 

294 "properties" 

295 ]["comments"]["items"]["required"], 

296 ) 

297 for comment in obj.comment_set.prefetch_related("user") 

298 ], 

299 "suggestions": lambda obj: [ 

300 self.backup_object( 

301 suggestion, 

302 self.component_schema["properties"]["units"]["items"][ 

303 "properties" 

304 ]["suggestions"]["items"]["required"], 

305 extras={ 

306 "votes": lambda obj: [ 

307 self.backup_object( 

308 vote, 

309 self.component_schema["properties"][ 

310 "units" 

311 ]["items"]["properties"]["suggestions"][ 

312 "items" 

313 ]["properties"]["votes"]["items"][ 

314 "required" 

315 ], 

316 ) 

317 for vote in obj.votes.through.objects.filter( 

318 suggestion=obj 

319 ).select_related("user") 

320 ], 

321 }, 

322 ) 

323 for suggestion in obj.suggestion_set.prefetch_related( 

324 "user" 

325 ) 

326 ], 

327 "checks": lambda obj: [ 

328 self.backup_object( 

329 check, 

330 self.component_schema["properties"]["units"]["items"][ 

331 "properties" 

332 ]["checks"]["items"]["required"], 

333 ) 

334 for check in obj.check_set.all() 

335 ], 

336 "labels": lambda obj: list( 

337 obj.labels.values_list("name", flat=True) 

338 ), 

339 }, 

340 ) 

341 for unit in Unit.objects.filter( 

342 translation__component=component 

343 ).iterator() 

344 ], 

345 "pending_unit_changes": [ 

346 self.backup_object( 

347 pending_unit_change, 

348 self.component_schema["properties"]["pending_unit_changes"][ 

349 "items" 

350 ]["required"], 

351 extras={ 

352 "unit_id_hash": lambda obj: obj.unit.checksum, 

353 "translation_id": lambda obj: obj.unit.translation_id, 

354 }, 

355 ) 

356 for pending_unit_change in PendingUnitChange.objects.for_component( 

357 component 

358 ) 

359 .prefetch_related("unit", "author") 

360 .iterator(2000) 

361 ], 

362 } 

363 # component category is not a required field 

364 if component.category: 

365 data["component"]["category"] = self.full_slug_without_project( 

366 component.category 

367 ) 

368 

369 data["screenshots"] = screenshots = [] 

370 for screenshot in Screenshot.objects.filter( 

371 translation__component=component 

372 ).prefetch_related("units"): 

373 screenshots.append( 

374 self.backup_object( 

375 screenshot, 

376 self.component_schema["properties"]["screenshots"]["items"][ 

377 "required" 

378 ], 

379 extras={ 

380 "units": lambda obj: [ 

381 hash_to_checksum(id_hash) 

382 for id_hash in obj.units.values_list("id_hash", flat=True) 

383 ], 

384 }, 

385 ) 

386 ) 

387 backupzip.write( 

388 os.path.join(settings.MEDIA_ROOT, screenshot.image.path), 

389 os.path.join("screenshots", os.path.basename(screenshot.image.name)), 

390 ) 

391 

392 validate_schema(data, "weblate-component.schema.json") 

393 self.backup_json( 

394 backupzip, 

395 data, 

396 f"{self.COMPONENTS_PREFIX}{self.full_slug_without_project(component)}.json", 

397 ) 

398 

399 # Store VCS repo in case it is present 

400 if component.is_repo_link: 

401 return 

402 

403 # Compact the repository 

404 with component.repository.lock: 

405 component.repository.compact() 

406 

407 # Actually perform the backup 

408 self.backup_dir( 

409 backupzip, 

410 component.full_path, 

411 f"{self.VCS_PREFIX}{self.full_slug_without_project(component)}", 

412 ) 

413 

414 @transaction.atomic 

415 def backup_project(self, project: Project) -> None: 

416 """Backup whole project.""" 

417 # Generate data 

418 self.backup_data(project) 

419 

420 self.generate_filename(project) 

421 part_name = f"{self.filename}.part" 

422 

423 # Create the zip with the content 

424 with ZipFile(part_name, "x") as backupzip: 

425 # Project data 

426 self.backup_json( 

427 backupzip, 

428 self.data, 

429 "weblate-backup.json", 

430 ) 

431 

432 # Translation memory, avoid using memory_db 

433 self.backup_json( 

434 backupzip, 

435 [ 

436 item.as_dict() 

437 for item in project.memory_set.using("default").iterator() 

438 ], 

439 "weblate-memory.json", 

440 ) 

441 

442 # Components 

443 for component in project.component_set.iterator(): 

444 self.backup_component(backupzip, component) 

445 

446 os.rename(part_name, self.filename) 

447 

448 def list_components(self, zipfile: ZipFile) -> list[str]: 

449 return [ 

450 name 

451 for name in zipfile.namelist() 

452 if name.startswith(self.COMPONENTS_PREFIX) 

453 ] 

454 

455 def load_data(self, zipfile: ZipFile) -> None: 

456 with zipfile.open("weblate-backup.json") as handle: 

457 self.data = json.load(handle) 

458 self.validate_data() 

459 self.timestamp = datetime.fromisoformat(self.data["metadata"]["timestamp"]) 

460 

461 def load_memory(self, zipfile: ZipFile) -> dict: 

462 with zipfile.open("weblate-memory.json") as handle: 

463 data = json.load(handle) 

464 validate_schema(data, "weblate-memory.schema.json") 

465 return data 

466 

467 def load_component( 

468 self, 

469 zipfile: ZipFile, 

470 filename: str, 

471 *, 

472 skip_linked: bool = False, 

473 do_restore: bool = False, 

474 ) -> bool: 

475 with zipfile.open(filename) as handle: 

476 data = json.load(handle) 

477 validate_schema(data, "weblate-component.schema.json") 

478 if skip_linked and data["component"]["repo"].startswith("weblate:"): 

479 return False 

480 if data["component"]["vcs"] not in VCS_REGISTRY: 

481 msg = f"Component {data['component']['name']} uses unsupported VCS: {data['component']['vcs']}" 

482 raise ValueError(msg) 

483 # Validate translations have unique languages 

484 languages = defaultdict(list) 

485 for item in data["translations"]: 

486 language = self.import_language(item["language_code"]) 

487 languages[language.code].append(item["language_code"]) 

488 

489 for code, values in languages.items(): 

490 if len(values) > 1: 

491 msg = f"Several languages from backup map to single language on this server {values} -> {code}" 

492 raise ValueError(msg) 

493 

494 if do_restore: 

495 self.restore_component(zipfile, data) 

496 return True 

497 

498 def load_components(self, zipfile: ZipFile, *, do_restore: bool = False) -> None: 

499 pending: list[str] = [] 

500 for component in self.list_components(zipfile): 

501 processed = self.load_component( 

502 zipfile, component, skip_linked=True, do_restore=do_restore 

503 ) 

504 if not processed: 

505 pending.append(component) 

506 for component in pending: 

507 self.load_component( 

508 zipfile, component, skip_linked=False, do_restore=do_restore 

509 ) 

510 

511 def validate(self) -> None: 

512 if not self.supports_restore: 

513 msg = "Restore is not supported on this database." 

514 raise ValueError(msg) 

515 input_file = self.filename or self.fileio 

516 if input_file is None: 

517 msg = "Can not validate None file." 

518 raise TypeError(msg) 

519 with ZipFile(input_file, "r") as zipfile: 

520 names = zipfile.namelist() 

521 if len(names) != len(set(names)): 

522 msg = "The zip file contains duplicate files. Please generate a new backup with a newer version of Weblate." 

523 raise ValueError(msg) 

524 self.load_data(zipfile) 

525 self.load_memory(zipfile) 

526 self.load_components(zipfile) 

527 for name in zipfile.namelist(): 

528 validate_filename(name) 

529 

530 def restore_unit( 

531 self, 

532 item: dict, 

533 translation_lookup: dict[int, Translation], 

534 source_unit_lookup: dict[int, Unit] | None = None, 

535 ) -> Unit: 

536 kwargs = item.copy() 

537 for skip in ("labels", "comments", "suggestions", "checks", "pending"): 

538 kwargs.pop(skip, None) 

539 kwargs["id_hash"] = checksum_to_hash(kwargs["id_hash"]) 

540 kwargs["translation_id"] = translation_lookup[kwargs["translation_id"]].id 

541 unit = Unit(**kwargs) 

542 unit.import_data = item 

543 if source_unit_lookup is not None: 

544 unit.source_unit = source_unit_lookup[item["id_hash"]] 

545 return unit 

546 

547 def restore_user(self, username: str) -> User: 

548 if not self.user_cache: 

549 self.user_cache[settings.ANONYMOUS_USER_NAME] = get_anonymous() 

550 if username not in self.user_cache: 

551 try: 

552 self.user_cache[username] = User.objects.get(username=username) 

553 except User.DoesNotExist: 

554 # Fallback to anonymous? 

555 self.user_cache[username] = self.user_cache[ 

556 settings.ANONYMOUS_USER_NAME 

557 ] 

558 

559 return self.user_cache[username] 

560 

561 def restore_with_user( 

562 self, data: dict[str, Any], field: str = "user", remove: str | None = None 

563 ) -> dict[str, Any]: 

564 data = data.copy() 

565 if remove is not None: 

566 data.pop(remove) 

567 data[field] = self.restore_user(data[field]) 

568 return data 

569 

570 def restore_users(self, usernames: list[str]) -> list[User]: 

571 users = [] 

572 for username in usernames: 

573 user = self.restore_user(username) 

574 if user.username == settings.ANONYMOUS_USER_NAME: 

575 continue 

576 users.append(user) 

577 return users 

578 

579 @staticmethod 

580 def get_items_from_cache(cache: dict[str, Any], keys: list[str]) -> list: 

581 return [value for key in keys if (value := cache.get(key))] 

582 

583 def restore_team(self, team: dict) -> None: 

584 if team["name"] == "Administration": 

585 group = Group.objects.get(name=team["name"], defining_project=self.project) 

586 else: 

587 group = Group(name=team["name"], defining_project=self.project) 

588 group = Group.objects.bulk_create([group])[0] 

589 

590 group.language_selection = team["language_selection"] 

591 group.enforced_2fa = team["enforced_2fa"] 

592 

593 group.roles.set(self.get_items_from_cache(self.roles_cache, team["roles"])) 

594 group.components.set( 

595 self.get_items_from_cache(self.components_cache, team["components"]) 

596 ) 

597 group.languages.set( 

598 self.get_items_from_cache(self.languages_cache, team["languages"]) 

599 ) 

600 group.admins.set(self.restore_users(team["admins"])) 

601 group.user_set.set(self.restore_users(team["members"])) 

602 

603 autogroups = [ 

604 AutoGroup(match=match, group=group) for match in team["autogroups"] 

605 ] 

606 AutoGroup.objects.bulk_create(autogroups) 

607 

608 def restore_teams(self, data: list[dict]) -> None: 

609 self.roles_cache = {r.name: r for r in Role.objects.all()} 

610 self.create_language_cache() 

611 for team in data: 

612 self.restore_team(team) 

613 

614 def restore_pending_unit_changes( 

615 self, 

616 data: dict, 

617 *, 

618 translation_lookup: dict, 

619 source_units: list[Unit], 

620 units: list[Unit], 

621 ) -> None: 

622 if "pending_unit_changes" in data: 

623 all_units: dict[int, dict[int, Unit]] = defaultdict(dict) 

624 for unit in chain(source_units, units): 

625 all_units[unit.translation.id][unit.checksum] = unit 

626 

627 pending_unit_changes = [] 

628 for item in data["pending_unit_changes"]: 

629 new_translation = translation_lookup[item["translation_id"]] 

630 unit = all_units[new_translation.id][item["unit_id_hash"]] 

631 pending_unit_changes.append( 

632 PendingUnitChange( 

633 unit=unit, 

634 author=self.restore_user(item["author"]), 

635 target=item["target"], 

636 explanation=item["explanation"], 

637 source_unit_explanation=item["source_unit_explanation"], 

638 timestamp=item["timestamp"], 

639 add_unit=item["add_unit"], 

640 state=item["state"], 

641 ) 

642 ) 

643 else: 

644 pending_unit_changes = [ 

645 PendingUnitChange( 

646 unit=unit, 

647 author=unit.get_last_content_change()[0], 

648 target=unit.target, 

649 explanation=unit.explanation, 

650 source_unit_explanation=unit.source_unit.explanation, 

651 state=unit.state, 

652 add_unit=unit.details.get("add_unit", False), 

653 ) 

654 for unit in chain(source_units, units) 

655 if unit.import_data.get("pending") 

656 ] 

657 

658 if pending_unit_changes: 

659 PendingUnitChange.objects.bulk_create(pending_unit_changes) 

660 

661 def restore_component(self, zipfile: ZipFile, data: dict) -> None: # noqa: C901 

662 if self.project is None: 

663 raise TypeError 

664 kwargs = data["component"].copy() 

665 source_language = kwargs["source_language"] = self.import_language( 

666 kwargs["source_language"] 

667 ) 

668 

669 # Fixup linked components 

670 if kwargs["repo"].startswith("weblate:"): 

671 old_slug = f"weblate://{self.data['project']['slug']}/" 

672 new_slug = f"weblate://{self.project.slug}/" 

673 kwargs["repo"] = kwargs["repo"].replace(old_slug, new_slug) 

674 # Update linked_component attribute 

675 if kwargs["repo"].startswith(new_slug): 

676 kwargs["linked_component"] = self.components_cache[ 

677 kwargs["repo"].removeprefix(new_slug) 

678 ] 

679 

680 if "category" in kwargs: 

681 kwargs["category"] = self.categories_cache[kwargs["category"]] 

682 

683 component = Component(project=self.project, **kwargs) 

684 # Trigger pre_save to update git export URL 

685 pre_save.send( 

686 sender=component.__class__, 

687 instance=component, 

688 raw=False, 

689 using=None, 

690 update_fields=None, 

691 ) 

692 # Use bulk create to avoid triggering save() and any post_save signals 

693 component = Component.objects.bulk_create([component])[0] 

694 

695 # Create translations 

696 translations = [] 

697 source_translation_id = -1 

698 for item in data["translations"]: 

699 language = self.import_language(item["language_code"]) 

700 plurals = language.plural_set.filter(**item["plural"]) 

701 try: 

702 plural = plurals[0] 

703 except IndexError: 

704 if item["plural"]["source"] == Plural.SOURCE_DEFAULT: 

705 plural = language.plural 

706 elif item["plural"]["source"] in { 

707 Plural.SOURCE_MANUAL, 

708 Plural.SOURCE_GETTEXT, 

709 }: 

710 plural = language.plural_set.create(**item["plural"]) 

711 else: 

712 plural = language.plural_set.filter( 

713 source=item["plural"]["source"] 

714 )[0] 

715 translation = Translation( 

716 component=component, 

717 filename=item["filename"], 

718 language_code=item["language_code"], 

719 language=self.import_language(item["language_code"]), 

720 plural=plural, 

721 revision=item["revision"], 

722 ) 

723 translation.original_id = item["id"] 

724 if language == source_language: 

725 source_translation_id = item["id"] 

726 translations.append(translation) 

727 translations = Translation.objects.bulk_create(translations)