add sync tool (wip)

.gitignore

@@ -8,3 +8,4 @@ shared/static/shared/*.css
 webpack-stats.json
 .venv
 media/*
+git

contracts/management/commands/import_old_contracts.py

+import io
 import os
 
+from datetime import date, datetime
+
+import yaml
+
+from ...models import Contract, ContractFilingArea, ContractType
+
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from git import Repo
-from markdown import Markdown
 
 
 class Command(BaseCommand):
@@ -25,6 +31,210 @@ class Command(BaseCommand):
             type=str,
             help="Directory to store the cloned repository in",
         )
+        parser.add_argument(
+            "--existing",
+            action="store_true",
+            help="Use the existing storage directory, as long as it exists.",
+        )
+
+    def parse_index(
+        self,
+        contract_root: str,
+        open_file,
+    ) -> dict:
+        split_contents = open_file.read().split("---")
+
+        if len(split_contents) < 2:
+            self.stderr.write(
+                self.style.NOTICE(f"{contract_root} index does not have valid metadata.")
+            )
+            raise ValueError
+
+        yaml_source = split_contents[1]
+
+        try:
+            parsed_metadata = yaml.safe_load(io.StringIO(yaml_source))
+        except yaml.YAMLError as exc:
+            self.stderr.write(
+                self.style.NOTICE(f"Failed to parse {contract_root} metadata.")
+            )
+
+            raise ValueError from exc
+
+        return parsed_metadata
+
+    def assign_contract_metadata(
+        self,
+        contract: Contract,
+        metadata: dict,
+    ) -> None:
+        filing_area = None
+        types = []
+        contract_already_exists = False
+
+        for key, value in metadata.items():
+            key = key.strip()
+
+            if isinstance(value, str):
+                value = value.strip()
+
+            if key == "datum účinnosti" and isinstance(value, date):
+                contract.valid_start_date = value
+            elif key == "datum ukončení" and isinstance(value, date):
+                contract.valid_end_date = value
+            elif key == "title":
+                contract.name = value
+
+                if Contract.objects.filter(name=value).exists():
+                    contract_already_exists = True
+                    self.stdout.write(f"{contract.name} already exists.")
+
+                    break
+            elif key == "použité smluvní typy":
+                if not isinstance(value, list):
+                    continue
+
+                for type_name in value:
+                    if not isinstance(type_name, str):
+                        continue
+
+                    type_name = type_name.strip()
+
+                    try:
+                        type_instance = ContractType.objects.get(name=type_name)
+                    except ContractType.DoesNotExist:
+                        type_instance = ContractType(name=type_name)
+
+                    types.append(type_instance)
+            elif key == "předmět":
+                contract.summary = value
+            elif key == "stav":
+                pass  # TODO
+            elif key == "náklady":
+                if isinstance(value, int):
+                    if value <= 0:
+                        continue
+
+                    contract.cost_amount = value
+                    contract.cost_unit = contract.CostUnits.TOTAL
+                elif contract.name is not None:
+                    self.stdout.write(
+                        self.style.WARNING(f"Could not parse cost for contract {contract.name}.")
+                    )
+            elif key == "místo uložení":
+                try:
+                    filing_area = ContractFilingArea.objects.get(name=value)
+                except ContractFilingArea.DoesNotExist:
+                    if isinstance(value, str):
+                        filing_area = ContractFilingArea(name=value)
+
+        if not contract_already_exists:
+            if filing_area is not None:
+                filing_area.save()
+
+            for type_ in types:
+                type_.save()
+
+            # Save primary key first
+            contract.save()
+
+            contract.filing_area = filing_area
+            contract.types.set(types)
+            contract.save()
+
+    def import_contract_from_files(
+        self,
+        contract_root: str,
+        files: list[str],
+        valid_start_date: datetime
+    ) -> None:
+        for file_ in files:
+            with open(
+                os.path.join(
+                    contract_root,
+                    file_,
+                ),
+                "r"
+            ) as open_file:
+                contract = Contract()
+
+                if file_ == "index.html":
+                    try:
+                        metadata = self.parse_index(contract_root, open_file)
+                    except ValueError:
+                        continue
+
+                    if metadata is None:
+                        continue
+
+                    self.assign_contract_metadata(contract, metadata)
+                elif file_.endswith(".pdf"):
+                    continue
+
+            contract.save()
+
+    def import_all_contracts(self, git_dir) -> None:
+        year_root = os.path.join(
+            git_dir,
+            "smlouvy",
+        )
+
+        saved_count = 0
+
+        for year_directory in os.listdir(year_root):
+            if int(year_directory) == 0:
+                continue  # Out of range, TODO
+
+            month_root = os.path.join(
+                year_root,
+                year_directory,
+            )
+
+            for month_directory in os.listdir(month_root):
+                day_root = os.path.join(
+                    month_root,
+                    month_directory,
+                )
+
+                for day_directory in os.listdir(day_root):
+                    contract_root = os.path.join(
+                        git_dir,
+                        "smlouvy",
+                        year_directory,
+                        month_directory,
+                        day_directory,
+                    )
+
+                    for contract_directory in os.listdir(contract_root):
+                        this_contract_directory = os.path.join(
+                            contract_root,
+                            contract_directory,
+                        )
+
+                        if not os.path.isdir(this_contract_directory):
+                            self.stderr.write(
+                                self.style.NOTICE(
+                                    f"{this_contract_directory} is not a directory, skipping."
+                                )
+                            )
+                            continue
+
+                        valid_start_date = datetime(
+                            year=int(year_directory),
+                            month=int(month_directory),
+                            day=int(day_directory)
+                        )
+
+                        self.import_contract_from_files(
+                            this_contract_directory,
+                            os.listdir(this_contract_directory),
+                            valid_start_date,
+                        )
+                        saved_count += 1
+
+        self.stdout.write(
+            self.style.SUCCESS(f"Saved {saved_count} contracts.")
+        )
 
     def handle(self, *args, **options):
         git_dir = os.path.join(
@@ -36,15 +246,31 @@ class Command(BaseCommand):
             )
         )
 
-        Repo.clone_from(
-            options["repo_url"],
-            git_dir,
-            branch=options["branch"],
-        )
+        if os.path.exists(git_dir):
+            if not options["existing"]:
+                self.stderr.write(
+                    self.style.ERROR(
+                        f"Temporary git storage directory ({git_dir}) already exists. "
+                        "As it could contain other data, it will not be removed. Please "
+                        "remove it manually and try again."
+                    )
+                )
 
-        md = markdown.Markdown(extensions=['meta'])
-        parsed_metadata = md.convert("").Meta
+                return
+            else:
+                self.stdout.write("Using existing git storage directory.")
+        else:
+            Repo.clone_from(
+                options["repo_url"],
+                git_dir,
+                branch=options["branch"],
+            )
+
+            self.stdout.write("Cloning repository.")
 
-        
+        self.stdout.write("\n")
+        self.import_all_contracts(git_dir)
 
-        self.stdout.write("\nGit repository sync complete.")
+        self.stdout.write(
+            self.style.SUCCESS("\nGit repository sync complete.")
+        )

contracts/migrations/0050_alter_contractfilingarea_name_and_more.py

+# Generated by Django 4.1.4 on 2023-04-20 09:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contracts', '0049_alter_contract_options'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='contractfilingarea',
+            name='name',
+            field=models.CharField(max_length=128, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contractissue',
+            name='name',
+            field=models.CharField(max_length=128, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contracttype',
+            name='name',
+            field=models.CharField(max_length=128, verbose_name='Jméno'),
+        ),
+    ]

contracts/migrations/0051_alter_contract_cost_unit_other_and_more.py

+# Generated by Django 4.1.4 on 2023-04-20 09:53
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contracts', '0050_alter_contractfilingarea_name_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='contract',
+            name='cost_unit_other',
+            field=models.CharField(blank=True, help_text='Je nutno vyplnit v případě, že máš vybranou možnost "jiné" v jednotce nákladů.', max_length=256, null=True, verbose_name='Jednotka nákladů (jiné)'),
+        ),
+        migrations.AlterField(
+            model_name='contract',
+            name='id_number',
+            field=models.CharField(blank=True, max_length=256, null=True, verbose_name='Identifikační číslo'),
+        ),
+        migrations.AlterField(
+            model_name='contract',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Název'),
+        ),
+        migrations.AlterField(
+            model_name='contractapproval',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contractee',
+            name='department',
+            field=models.CharField(blank=True, max_length=256, null=True, verbose_name='Organizační složka'),
+        ),
+        migrations.AlterField(
+            model_name='contractfile',
+            name='name',
+            field=models.CharField(blank=True, max_length=256, null=True, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contractfilingarea',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contractintent',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contractissue',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='contracttype',
+            name='name',
+            field=models.CharField(max_length=256, verbose_name='Jméno'),
+        ),
+        migrations.AlterField(
+            model_name='signee',
+            name='department',
+            field=models.CharField(blank=True, max_length=256, null=True, verbose_name='Organizační složka'),
+        ),
+    ]

contracts/models.py

@@ -188,7 +188,7 @@ class Signee(
     )  # WARNING: Legal entity status dependent!
 
     department = models.CharField(
-        max_length=128,
+        max_length=256,
         blank=True,
         null=True,
         verbose_name="Organizační složka",
@@ -341,7 +341,7 @@ class Contractee(
     )
 
     department = models.CharField(
-        max_length=128,
+        max_length=256,
         blank=True,
         null=True,
         verbose_name="Organizační složka",
@@ -370,7 +370,7 @@ class Contractee(
 
 class ContractType(ContractCountMixin, NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=32,
+        max_length=256,
         verbose_name="Jméno",
     )
 
@@ -387,7 +387,7 @@ class ContractType(ContractCountMixin, NameStrMixin, models.Model):
 
 class ContractIssue(ContractCountMixin, NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=32,
+        max_length=256,
         verbose_name="Jméno",
     )
 
@@ -404,7 +404,7 @@ class ContractIssue(ContractCountMixin, NameStrMixin, models.Model):
 
 class ContractFilingArea(ContractCountMixin, NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=32,
+        max_length=256,
         verbose_name="Jméno",
     )
 
@@ -490,12 +490,12 @@ class Contract(NameStrMixin, models.Model):
     # END Approval fields
 
     name = models.CharField(
-        max_length=128,
+        max_length=256,
         verbose_name="Název",
     )
 
     id_number = models.CharField(
-        max_length=128,
+        max_length=256,
         blank=True,
         null=True,
         verbose_name="Identifikační číslo",
@@ -622,7 +622,7 @@ class Contract(NameStrMixin, models.Model):
     )
 
     cost_unit_other = models.CharField(
-        max_length=128,
+        max_length=256,
         verbose_name="Jednotka nákladů (jiné)",
         help_text='Je nutno vyplnit v případě, že máš vybranou možnost "jiné" v jednotce nákladů.',
         blank=True,
@@ -825,7 +825,7 @@ def get_contract_file_loaction(instance, filename):
 
 class ContractFile(NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=128,
+        max_length=256,
         blank=True,
         null=True,
         verbose_name="Jméno",
@@ -995,7 +995,7 @@ def signing_parties_post_save_update_dates(sender, instance, *args, **kwargs) ->
 
 class ContractApproval(NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=128,
+        max_length=256,
         verbose_name="Jméno",
     )
 
@@ -1024,7 +1024,7 @@ class ContractApproval(NameStrMixin, models.Model):
 
 class ContractIntent(NameStrMixin, models.Model):
     name = models.CharField(
-        max_length=128,
+        max_length=256,
         verbose_name="Jméno",
     )
 

requirements/base.txt

@@ -18,3 +18,4 @@ django-guardian==2.4.0
 GitPython==3.1.31
 Markdown==3.4.3
 PyJWT==2.6.0
+PyYAML==6.0