Funkwhale layout ¶

tree -L 6 ¶

See also

https://dev.funkwhale.audio/funkwhale/funkwhale

179 directories
741 files

           .
├── api
│   ├── compose
│   │   └── django
│   │       ├── daphne.sh
│   │       ├── dev-entrypoint.sh
│   │       └── entrypoint.sh
│   ├── config
│   │   ├── api_urls.py
│   │   ├── asgi.py
│   │   ├── __init__.py
│   │   ├── routing.py
│   │   ├── settings
│   │   │   ├── common.py
│   │   │   ├── __init__.py
│   │   │   ├── local.py
│   │   │   └── production.py
│   │   ├── spa_urls.py
│   │   └── urls.py
│   ├── Dockerfile
│   ├── funkwhale_api
│   │   ├── activity
│   │   │   ├── apps.py
│   │   │   ├── __init__.py
│   │   │   ├── record.py
│   │   │   ├── serializers.py
│   │   │   ├── utils.py
│   │   │   └── views.py
│   │   ├── common
│   │   │   ├── admin.py
│   │   │   ├── apps.py
│   │   │   ├── authentication.py
│   │   │   ├── auth.py
│   │   │   ├── channels.py
│   │   │   ├── consumers.py
│   │   │   ├── decorators.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── factories.py
│   │   │   ├── fields.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── management
│   │   │   │   ├── commands
│   │   │   │   │   ├── __init__.py
│   │   │   │   │   ├── makemigrations.py
│   │   │   │   │   ├── migrate.py
│   │   │   │   │   └── script.py
│   │   │   │   └── __init__.py
│   │   │   ├── middleware.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_mutation.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── mutations.py
│   │   │   ├── pagination.py
│   │   │   ├── permissions.py
│   │   │   ├── preferences.py
│   │   │   ├── scripts
│   │   │   │   ├── create_actors.py
│   │   │   │   ├── create_image_variations.py
│   │   │   │   ├── delete_pre_017_federated_uploads.py
│   │   │   │   ├── django_permissions_to_user_permissions.py
│   │   │   │   ├── __init__.py
│   │   │   │   ├── migrate_to_user_libraries.py
│   │   │   │   └── test.py
│   │   │   ├── search.py
│   │   │   ├── serializers.py
│   │   │   ├── session.py
│   │   │   ├── signals.py
│   │   │   ├── storage.py
│   │   │   ├── tasks.py
│   │   │   ├── utils.py
│   │   │   ├── validators.py
│   │   │   └── views.py
│   │   ├── contrib
│   │   │   ├── __init__.py
│   │   │   └── sites
│   │   │       ├── __init__.py
│   │   │       └── migrations
│   │   │           ├── 0001_initial.py
│   │   │           ├── 0002_set_site_domain_and_name.py
│   │   │           ├── 0003_auto_20171214_2205.py
│   │   │           └── __init__.py
│   │   ├── factories.py
│   │   ├── favorites
│   │   │   ├── activities.py
│   │   │   ├── admin.py
│   │   │   ├── consumers.py
│   │   │   ├── factories.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── serializers.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── federation
│   │   │   ├── activity.py
│   │   │   ├── actors.py
│   │   │   ├── admin.py
│   │   │   ├── api_serializers.py
│   │   │   ├── api_urls.py
│   │   │   ├── api_views.py
│   │   │   ├── authentication.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── exceptions.py
│   │   │   ├── factories.py
│   │   │   ├── fields.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── keys.py
│   │   │   ├── library.py
│   │   │   ├── management
│   │   │   │   ├── commands
│   │   │   │   │   ├── fix_federation_ids.py
│   │   │   │   │   └── __init__.py
│   │   │   │   └── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_auto_20180403_1620.py
│   │   │   │   ├── 0003_auto_20180407_1010.py
│   │   │   │   ├── 0004_auto_20180410_2025.py
│   │   │   │   ├── 0005_auto_20180413_1723.py
│   │   │   │   ├── 0006_auto_20180521_1702.py
│   │   │   │   ├── 0007_auto_20180807_1748.py
│   │   │   │   ├── 0008_auto_20180807_1748.py
│   │   │   │   ├── 0009_auto_20180822_1956.py
│   │   │   │   ├── 0010_auto_20180904_2011.py
│   │   │   │   ├── 0011_auto_20180910_1902.py
│   │   │   │   ├── 0012_auto_20180920_1803.py
│   │   │   │   ├── 0013_auto_20181226_1935.py
│   │   │   │   ├── 0014_auto_20181205_0958.py
│   │   │   │   ├── 0015_populate_domains.py
│   │   │   │   ├── 0016_auto_20181227_1605.py
│   │   │   │   ├── 0017_auto_20190130_0926.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── parsers.py
│   │   │   ├── renderers.py
│   │   │   ├── routes.py
│   │   │   ├── serializers.py
│   │   │   ├── signing.py
│   │   │   ├── tasks.py
│   │   │   ├── urls.py
│   │   │   ├── utils.py
│   │   │   ├── views.py
│   │   │   └── webfinger.py
│   │   ├── history
│   │   │   ├── activities.py
│   │   │   ├── admin.py
│   │   │   ├── factories.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_auto_20180325_1433.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── serializers.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── __init__.py
│   │   ├── instance
│   │   │   ├── consumers.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── __init__.py
│   │   │   ├── nodeinfo.py
│   │   │   ├── stats.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── manage
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── serializers.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── moderation
│   │   │   ├── admin.py
│   │   │   ├── factories.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_auto_20190213_0927.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── serializers.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── music
│   │   │   ├── admin.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── factories.py
│   │   │   ├── fake_data.py
│   │   │   ├── filters.py
│   │   │   ├── importers.py
│   │   │   ├── __init__.py
│   │   │   ├── licenses.py
│   │   │   ├── lyrics.py
│   │   │   ├── management
│   │   │   │   ├── commands
│   │   │   │   │   ├── check_licenses.py
│   │   │   │   │   ├── fix_uploads.py
│   │   │   │   │   ├── import_files.py
│   │   │   │   │   └── __init__.py
│   │   │   │   └── __init__.py
│   │   │   ├── metadata.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_auto_20151215_1645.py
│   │   │   │   ├── 0003_auto_20151222_2233.py
│   │   │   │   ├── 0004_track_tags.py
│   │   │   │   ├── 0005_deduplicate.py
│   │   │   │   ├── 0006_unique_mbid.py
│   │   │   │   ├── 0007_track_position.py
│   │   │   │   ├── 0008_auto_20160529_1456.py
│   │   │   │   ├── 0009_auto_20160920_1614.py
│   │   │   │   ├── 0010_auto_20160920_1742.py
│   │   │   │   ├── 0011_rename_files.py
│   │   │   │   ├── 0012_auto_20161122_1905.py
│   │   │   │   ├── 0013_auto_20171213_2211.py
│   │   │   │   ├── 0014_importjob_track_file.py
│   │   │   │   ├── 0015_bind_track_file_to_import_job.py
│   │   │   │   ├── 0016_trackfile_acoustid_track_id.py
│   │   │   │   ├── 0017_auto_20171227_1728.py
│   │   │   │   ├── 0018_auto_20180218_1554.py
│   │   │   │   ├── 0019_populate_mimetypes.py
│   │   │   │   ├── 0020_importbatch_status.py
│   │   │   │   ├── 0021_populate_batch_status.py
│   │   │   │   ├── 0022_importbatch_import_request.py
│   │   │   │   ├── 0023_auto_20180407_1010.py
│   │   │   │   ├── 0024_populate_uuid.py
│   │   │   │   ├── 0025_auto_20180419_2023.py
│   │   │   │   ├── 0026_trackfile_accessed_date.py
│   │   │   │   ├── 0027_auto_20180515_1808.py
│   │   │   │   ├── 0028_importjob_replace_if_duplicate.py
│   │   │   │   ├── 0029_auto_20180807_1748.py
│   │   │   │   ├── 0030_auto_20180825_1411.py
│   │   │   │   ├── 0031_auto_20180914_2007.py
│   │   │   │   ├── 0032_track_file_to_upload.py
│   │   │   │   ├── 0033_auto_20181023_1837.py
│   │   │   │   ├── 0034_auto_20181127_0325.py
│   │   │   │   ├── 0035_auto_20181203_1515.py
│   │   │   │   ├── 0036_track_disc_number.py
│   │   │   │   ├── 0037_auto_20190103_1757.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── mutations.py
│   │   │   ├── serializers.py
│   │   │   ├── signals.py
│   │   │   ├── spa_views.py
│   │   │   ├── tasks.py
│   │   │   ├── utils.py
│   │   │   └── views.py
│   │   ├── musicbrainz
│   │   │   ├── client.py
│   │   │   ├── __init__.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── playlists
│   │   │   ├── admin.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── factories.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_auto_20180316_2217.py
│   │   │   │   ├── 0003_auto_20180319_1214.py
│   │   │   │   ├── 0004_auto_20180320_1713.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── serializers.py
│   │   │   └── views.py
│   │   ├── providers
│   │   │   ├── __init__.py
│   │   │   └── urls.py
│   │   ├── radios
│   │   │   ├── admin.py
│   │   │   ├── factories.py
│   │   │   ├── filtersets.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   ├── 0002_radiosession_session_key.py
│   │   │   │   ├── 0003_auto_20160521_1708.py
│   │   │   │   ├── 0004_auto_20180107_1813.py
│   │   │   │   └── __init__.py
│   │   │   ├── models.py
│   │   │   ├── radios.py
│   │   │   ├── registries.py
│   │   │   ├── serializers.py
│   │   │   ├── urls.py
│   │   │   └── views.py
│   │   ├── requests
│   │   │   ├── __init__.py
│   │   │   ├── migrations
│   │   │   │   ├── 0001_initial.py
│   │   │   │   └── __init__.py
│   │   │   └── models.py
│   │   ├── static
│   │   │   ├── css
│   │   │   │   └── project.css
│   │   │   ├── fonts
│   │   │   ├── images
│   │   │   │   └── favicon.ico
│   │   │   ├── js
│   │   │   │   └── project.js
│   │   │   ├── music
│   │   │   │   └── sample1.ogg
│   │   │   └── sass
│   │   │       └── project.scss
│   │   ├── subsonic
│   │   │   ├── authentication.py
│   │   │   ├── dynamic_preferences_registry.py
│   │   │   ├── filters.py
│   │   │   ├── __init__.py
│   │   │   ├── negotiation.py
│   │   │   ├── renderers.py
│   │   │   ├── serializers.py
│   │   │   └── views.py
│   │   ├── taskapp
│   │   │   ├── celery.py
│   │   │   └── __init__.py
│   │   ├── templates
│   │   │   ├── account
│   │   │   │   └── email
│   │   │   │       └── email_confirmation_message.txt
│   │   │   └── registration
│   │   │       └── password_reset_email.html
│   │   └── users
│   │       ├── adapters.py
│   │       ├── admin.py
│   │       ├── api_urls.py
│   │       ├── auth_backends.py
│   │       ├── dynamic_preferences_registry.py
│   │       ├── factories.py
│   │       ├── __init__.py
│   │       ├── middleware.py
│   │       ├── migrations
│   │       │   ├── 0001_initial.py
│   │       │   ├── 0002_auto_20171214_2205.py
│   │       │   ├── 0003_auto_20171226_1357.py
│   │       │   ├── 0004_user_privacy_level.py
│   │       │   ├── 0005_user_subsonic_api_token.py
│   │       │   ├── 0006_auto_20180517_2324.py
│   │       │   ├── 0007_auto_20180524_2009.py
│   │       │   ├── 0008_auto_20180617_1531.py
│   │       │   ├── 0009_auto_20180619_2024.py
│   │       │   ├── 0010_user_avatar.py
│   │       │   ├── 0011_auto_20180721_1317.py
│   │       │   ├── 0012_user_upload_quota.py
│   │       │   ├── 0013_auto_20181206_1008.py
│   │       │   └── __init__.py
│   │       ├── models.py
│   │       ├── permissions.py
│   │       ├── rest_auth_urls.py
│   │       ├── serializers.py
│   │       └── views.py
│   ├── install_os_dependencies.sh
│   ├── manage.py
│   ├── requirements
│   │   ├── base.txt
│   │   ├── local.txt
│   │   └── test.txt
│   ├── requirements.apt
│   ├── requirements.pac
│   ├── requirements.txt
│   ├── setup.cfg
│   └── tests
│       ├── activity
│       │   ├── __init__.py
│       │   ├── test_record.py
│       │   ├── test_serializers.py
│       │   ├── test_utils.py
│       │   └── test_views.py
│       ├── channels
│       │   ├── __init__.py
│       │   ├── test_auth.py
│       │   └── test_consumers.py
│       ├── common
│       │   ├── exif.jpg
│       │   ├── __init__.py
│       │   ├── test_decorators.py
│       │   ├── test_fields.py
│       │   ├── test_filters.py
│       │   ├── test_middleware.py
│       │   ├── test_models.py
│       │   ├── test_mutations.py
│       │   ├── test_permissions.py
│       │   ├── test_preferences.py
│       │   ├── test_scripts.py
│       │   ├── test_search.py
│       │   ├── test_serializers.py
│       │   ├── test_session.py
│       │   ├── test_tasks.py
│       │   ├── test_utils.py
│       │   └── test_views.py
│       ├── conftest.py
│       ├── data
│       │   └── youtube.py
│       ├── favorites
│       │   ├── __init__.py
│       │   ├── test_activity.py
│       │   ├── test_favorites.py
│       │   ├── test_filters.py
│       │   └── test_views.py
│       ├── federation
│       │   ├── __init__.py
│       │   ├── test_activity.py
│       │   ├── test_actors.py
│       │   ├── test_api_filters.py
│       │   ├── test_api_serializers.py
│       │   ├── test_api_views.py
│       │   ├── test_authentication.py
│       │   ├── test_commands.py
│       │   ├── test_keys.py
│       │   ├── test_migrations.py
│       │   ├── test_models.py
│       │   ├── test_routes.py
│       │   ├── test_serializers.py
│       │   ├── test_signing.py
│       │   ├── test_tasks.py
│       │   ├── test_utils.py
│       │   ├── test_views.py
│       │   └── test_webfinger.py
│       ├── files
│       │   ├── dummy_file.ogg
│       │   └── utf8-éà◌.ogg
│       ├── history
│       │   ├── __init__.py
│       │   ├── test_activity.py
│       │   ├── test_filters.py
│       │   ├── test_history.py
│       │   └── test_views.py
│       ├── __init__.py
│       ├── instance
│       │   ├── __init__.py
│       │   ├── test_nodeinfo.py
│       │   ├── test_preferences.py
│       │   ├── test_stats.py
│       │   └── test_views.py
│       ├── manage
│       │   ├── __init__.py
│       │   ├── test_filters.py
│       │   ├── test_serializers.py
│       │   └── test_views.py
│       ├── moderation
│       │   ├── __init__.py
│       │   ├── test_filters.py
│       │   ├── test_serializers.py
│       │   └── test_views.py
│       ├── music
│       │   ├── conftest.py
│       │   ├── cover.jpg
│       │   ├── cover.png
│       │   ├── __init__.py
│       │   ├── licenses.json
│       │   ├── sample.flac
│       │   ├── test_activity.py
│       │   ├── test_api.py
│       │   ├── test_commands.py
│       │   ├── test_filters.py
│       │   ├── test_import.py
│       │   ├── test_licenses.py
│       │   ├── test_lyrics.py
│       │   ├── test_metadata.py
│       │   ├── test_models.py
│       │   ├── test.mp3
│       │   ├── test_music.py
│       │   ├── test_mutations.py
│       │   ├── test.ogg
│       │   ├── test.opus
│       │   ├── test_serializers.py
│       │   ├── test_spa_views.py
│       │   ├── test_tasks.py
│       │   ├── test_theora.ogg
│       │   ├── test_utils.py
│       │   ├── test_views.py
│       │   ├── test_works.py
│       │   ├── with_cover.ogg
│       │   └── with_other_picture.mp3
│       ├── musicbrainz
│       │   ├── conftest.py
│       │   ├── __init__.py
│       │   ├── test_api.py
│       │   └── test_cache.py
│       ├── playlists
│       │   ├── __init__.py
│       │   ├── test_models.py
│       │   ├── test_serializers.py
│       │   └── test_views.py
│       ├── radios
│       │   ├── __init__.py
│       │   ├── test_api.py
│       │   └── test_radios.py
│       ├── subsonic
│       │   ├── test_authentication.py
│       │   ├── test_renderers.py
│       │   ├── test_serializers.py
│       │   └── test_views.py
│       ├── test_import_audio_file.py
│       ├── test_jwt_querystring.py
│       ├── test_tasks.py
│       └── users
│           ├── __init__.py
│           ├── test_activity.py
│           ├── test_admin.py
│           ├── test_jwt.py
│           ├── test_ldap.py
│           ├── test_middleware.py
│           ├── test_models.py
│           ├── test_permissions.py
│           └── test_views.py
├── archi.txt
├── CHANGELOG
├── changes
│   ├── changelog.d
│   │   ├── 356.bugfix
│   │   ├── 356.feature
│   │   ├── 578.enhancement
│   │   ├── 701.feature
│   │   ├── 702.bugfix
│   │   ├── 715.enhancement
│   │   ├── 722.bugfix
│   │   ├── 725.enhancement
│   │   ├── buttons.enhancement
│   │   ├── embed-wizard.enhancement
│   │   ├── similar-radio.enhancement
│   │   └── system-actor.enhancement
│   ├── __init__.py
│   ├── notes.rst
│   └── template.rst
├── CONTRIBUTING.rst
├── CONTRIBUTORS.txt
├── demo
│   ├── env.sample
│   ├── README.md
│   └── setup.sh
├── deploy
│   ├── apache.conf
│   ├── docker-compose.yml
│   ├── docker.funkwhale_proxy.conf
│   ├── docker.nginx.template
│   ├── docker.proxy.template
│   ├── env.prod.sample
│   ├── FreeBSD
│   │   ├── funkwhale_beat
│   │   ├── funkwhale_server
│   │   ├── funkwhale_worker
│   │   └── README.md
│   ├── funkwhale-beat.service
│   ├── funkwhale_proxy.conf
│   ├── funkwhale-server.service
│   ├── funkwhale.target
│   ├── funkwhale-worker.service
│   └── nginx.template
├── dev.yml
├── docker
│   ├── nginx
│   │   ├── conf.dev
│   │   └── entrypoint.sh
│   ├── ssl
│   │   ├── test.crt
│   │   └── test.key
│   ├── traefik.toml
│   └── traefik.yml
├── docs
│   ├── api.rst
│   ├── architecture.rst
│   ├── build_docs.sh
│   ├── build_swagger.sh
│   ├── changelog.rst
│   ├── configuration.rst
│   ├── conf.py
│   ├── contributing.rst
│   ├── developers
│   │   ├── index.rst
│   │   └── subsonic.rst
│   ├── Dockerfile
│   ├── features.rst
│   ├── federation
│   │   └── index.rst
│   ├── importing-music.rst
│   ├── index.rst
│   ├── installation
│   │   ├── debian.rst
│   │   ├── docker.rst
│   │   ├── external_dependencies.rst
│   │   ├── index.rst
│   │   ├── ldap.rst
│   │   ├── non_amd64_architectures.rst
│   │   ├── optimization.rst
│   │   └── systemd.rst
│   ├── Makefile
│   ├── serve.py
│   ├── swagger.yml
│   ├── third-party.rst
│   ├── translators.rst
│   ├── troubleshooting.rst
│   ├── upgrading
│   │   ├── 0.17.rst
│   │   └── index.rst
│   └── users
│       ├── apps.rst
│       ├── django.rst
│       ├── index.rst
│       ├── tagging.rst
│       └── upload.rst
├── front
│   ├── babel.config.js
│   ├── Dockerfile
│   ├── locales
│   │   ├── app.pot
│   │   ├── ar
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── de
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── eo
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── es
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── eu
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── fr_FR
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── gl
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── it
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── nb_NO
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── nl
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── oc
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── pl
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── pt_BR
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── pt_PT
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   ├── ru
│   │   │   └── LC_MESSAGES
│   │   │       └── app.po
│   │   └── sv
│   │       └── LC_MESSAGES
│   │           └── app.po
│   ├── package.json
│   ├── public
│   │   ├── custom.css
│   │   ├── embed.html
│   │   ├── favicon.png
│   │   ├── index.html
│   │   └── settings.json
│   ├── scripts
│   │   ├── i18n-compile.sh
│   │   ├── i18n-extract.sh
│   │   └── i18n-weblate-to-origin.sh
│   ├── src
│   │   ├── App.vue
│   │   ├── assets
│   │   │   ├── audio
│   │   │   │   └── default-cover.png
│   │   │   ├── embed
│   │   │   │   └── default-cover.jpeg
│   │   │   └── logo
│   │   │       ├── License.md
│   │   │       ├── logo-full-500.png
│   │   │       ├── logo-full.png
│   │   │       ├── logo.png
│   │   │       ├── logos.png
│   │   │       ├── logo.svg
│   │   │       └── logo-with-text.svg
│   │   ├── audio
│   │   │   ├── backend.js
│   │   │   └── formats.js
│   │   ├── components
│   │   │   ├── About.vue
│   │   │   ├── admin
│   │   │   │   └── SettingsGroup.vue
│   │   │   ├── audio
│   │   │   │   ├── album
│   │   │   │   │   ├── Card.vue
│   │   │   │   │   └── Widget.vue
│   │   │   │   ├── artist
│   │   │   │   │   └── Card.vue
│   │   │   │   ├── EmbedWizard.vue
│   │   │   │   ├── PlayButton.vue
│   │   │   │   ├── Player.vue
│   │   │   │   ├── SearchBar.vue
│   │   │   │   ├── Search.vue
│   │   │   │   ├── track
│   │   │   │   │   ├── Row.vue
│   │   │   │   │   ├── Table.vue
│   │   │   │   │   └── Widget.vue
│   │   │   │   └── Track.vue
│   │   │   ├── auth
│   │   │   │   ├── Login.vue
│   │   │   │   ├── Logout.vue
│   │   │   │   ├── Profile.vue
│   │   │   │   ├── Settings.vue
│   │   │   │   ├── Signup.vue
│   │   │   │   └── SubsonicTokenForm.vue
│   │   │   ├── common
│   │   │   │   ├── ActionTable.vue
│   │   │   │   ├── ActorAvatar.vue
│   │   │   │   ├── ActorLink.vue
│   │   │   │   ├── AjaxButton.vue
│   │   │   │   ├── CopyInput.vue
│   │   │   │   ├── DangerousButton.vue
│   │   │   │   ├── Duration.vue
│   │   │   │   ├── EmptyState.vue
│   │   │   │   ├── HumanDate.vue
│   │   │   │   ├── Message.vue
│   │   │   │   ├── Tooltip.vue
│   │   │   │   ├── UserLink.vue
│   │   │   │   └── Username.vue
│   │   │   ├── favorites
│   │   │   │   ├── List.vue
│   │   │   │   └── TrackFavoriteIcon.vue
│   │   │   ├── federation
│   │   │   │   └── LibraryWidget.vue
│   │   │   ├── Footer.vue
│   │   │   ├── forms
│   │   │   │   └── PasswordInput.vue
│   │   │   ├── globals.js
│   │   │   ├── Home.vue
│   │   │   ├── instance
│   │   │   │   └── Stats.vue
│   │   │   ├── library
│   │   │   │   ├── Albums.vue
│   │   │   │   ├── Album.vue
│   │   │   │   ├── Artists.vue
│   │   │   │   ├── Artist.vue
│   │   │   │   ├── EditCard.vue
│   │   │   │   ├── EditDetail.vue
│   │   │   │   ├── EditForm.vue
│   │   │   │   ├── EditList.vue
│   │   │   │   ├── FileUpload.vue
│   │   │   │   ├── FileUploadWidget.vue
│   │   │   │   ├── Home.vue
│   │   │   │   ├── Library.vue
│   │   │   │   ├── radios
│   │   │   │   │   ├── Builder.vue
│   │   │   │   │   └── Filter.vue
│   │   │   │   ├── Radios.vue
│   │   │   │   ├── TrackBase.vue
│   │   │   │   ├── TrackDetail.vue
│   │   │   │   └── TrackEdit.vue
│   │   │   ├── Logo.vue
│   │   │   ├── manage
│   │   │   │   ├── library
│   │   │   │   │   └── EditsCardList.vue
│   │   │   │   ├── moderation
│   │   │   │   │   ├── AccountsTable.vue
│   │   │   │   │   ├── DomainsTable.vue
│   │   │   │   │   ├── InstancePolicyCard.vue
│   │   │   │   │   └── InstancePolicyForm.vue
│   │   │   │   └── users
│   │   │   │       ├── InvitationForm.vue
│   │   │   │       ├── InvitationsTable.vue
│   │   │   │       └── UsersTable.vue
│   │   │   ├── metadata
│   │   │   │   ├── ArtistCard.vue
│   │   │   │   ├── CardMixin.vue
│   │   │   │   ├── ReleaseCard.vue
│   │   │   │   └── Search.vue
│   │   │   ├── mixins
│   │   │   │   ├── Ordering.vue
│   │   │   │   ├── Pagination.vue
│   │   │   │   ├── SmartSearch.vue
│   │   │   │   └── Translations.vue
│   │   │   ├── moderation
│   │   │   │   └── FilterModal.vue
│   │   │   ├── notifications
│   │   │   │   └── NotificationRow.vue
│   │   │   ├── PageNotFound.vue
│   │   │   ├── Pagination.vue
│   │   │   ├── playlists
│   │   │   │   ├── CardList.vue
│   │   │   │   ├── Card.vue
│   │   │   │   ├── Editor.vue
│   │   │   │   ├── Form.vue
│   │   │   │   ├── PlaylistModal.vue
│   │   │   │   ├── TrackPlaylistIcon.vue
│   │   │   │   └── Widget.vue
│   │   │   ├── radios
│   │   │   │   ├── Button.vue
│   │   │   │   └── Card.vue
│   │   │   ├── semantic
│   │   │   │   └── Modal.vue
│   │   │   ├── ServiceMessages.vue
│   │   │   ├── SetInstanceModal.vue
│   │   │   ├── ShortcutsModal.vue
│   │   │   ├── Sidebar.vue
│   │   │   └── utils
│   │   │       └── global-events.vue
│   │   ├── edits.js
│   │   ├── EmbedFrame.vue
│   │   ├── embed.js
│   │   ├── filters.js
│   │   ├── locales.js
│   │   ├── lodash.js
│   │   ├── logging.js
│   │   ├── main.js
│   │   ├── router
│   │   │   └── index.js
│   │   ├── search.js
│   │   ├── semantic.js
│   │   ├── store
│   │   │   ├── auth.js
│   │   │   ├── favorites.js
│   │   │   ├── index.js
│   │   │   ├── instance.js
│   │   │   ├── moderation.js
│   │   │   ├── player.js
│   │   │   ├── playlists.js
│   │   │   ├── queue.js
│   │   │   ├── radios.js
│   │   │   └── ui.js
│   │   ├── style
│   │   │   ├── _main.scss
│   │   │   ├── _site.scss
│   │   │   └── vendor
│   │   │       └── _media.scss
│   │   ├── utils
│   │   │   ├── color.js
│   │   │   ├── time.js
│   │   │   └── url.js
│   │   ├── vendor
│   │   │   └── color-thief.js
│   │   └── views
│   │       ├── admin
│   │       │   ├── library
│   │       │   │   ├── Base.vue
│   │       │   │   └── EditsList.vue
│   │       │   ├── moderation
│   │       │   │   ├── AccountsDetail.vue
│   │       │   │   ├── AccountsList.vue
│   │       │   │   ├── Base.vue
│   │       │   │   ├── DomainsDetail.vue
│   │       │   │   └── DomainsList.vue
│   │       │   ├── Settings.vue
│   │       │   └── users
│   │       │       ├── Base.vue
│   │       │       ├── InvitationsList.vue
│   │       │       └── UsersList.vue
│   │       ├── auth
│   │       │   ├── EmailConfirm.vue
│   │       │   ├── PasswordResetConfirm.vue
│   │       │   └── PasswordReset.vue
│   │       ├── content
│   │       │   ├── Base.vue
│   │       │   ├── Home.vue
│   │       │   ├── libraries
│   │       │   │   ├── Card.vue
│   │       │   │   ├── DetailArea.vue
│   │       │   │   ├── DetailMixin.vue
│   │       │   │   ├── Detail.vue
│   │       │   │   ├── FilesTable.vue
│   │       │   │   ├── Files.vue
│   │       │   │   ├── Form.vue
│   │       │   │   ├── Home.vue
│   │       │   │   ├── Quota.vue
│   │       │   │   └── Upload.vue
│   │       │   └── remote
│   │       │       ├── Card.vue
│   │       │       ├── Home.vue
│   │       │       └── ScanForm.vue
│   │       ├── Notifications.vue
│   │       ├── playlists
│   │       │   ├── Detail.vue
│   │       │   └── List.vue
│   │       └── radios
│   │           └── Detail.vue
│   ├── stats.json
│   ├── tests
│   │   └── unit
│   │       ├── specs
│   │       │   ├── components
│   │       │   │   └── common.spec.js
│   │       │   ├── filters
│   │       │   │   └── filters.spec.js
│   │       │   ├── search.spec.js
│   │       │   └── store
│   │       │       ├── auth.spec.js
│   │       │       ├── favorites.spec.js
│   │       │       ├── instance.spec.js
│   │       │       ├── player.spec.js
│   │       │       ├── playlists.spec.js
│   │       │       ├── queue.spec.js
│   │       │       ├── radios.spec.js
│   │       │       └── ui.spec.js
│   │       └── utils.js
│   ├── vue.config.js
│   └── yarn.lock
├── LICENSE
├── pyproject.toml
├── README.rst
├── scripts
│   ├── clean-unused-artifacts.py
│   └── set-api-build-metadata.sh
└── TRANSLATORS.rst

179 directories, 742 files

          

Python modules ¶

Django Models ¶

           ls **/models.py | wc -l

          

           api/funkwhale_api/common/models.py
api/funkwhale_api/favorites/models.py
api/funkwhale_api/federation/models.py
api/funkwhale_api/history/models.py
api/funkwhale_api/moderation/models.py
api/funkwhale_api/music/models.py
api/funkwhale_api/playlists/models.py
api/funkwhale_api/radios/models.py
api/funkwhale_api/requests/models.py
api/funkwhale_api/users/models.py

          

api/funkwhale_api/common/models.py ¶

            import uuid

from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import JSONField
from django.db import models
from django.db import transaction
from django.urls import reverse
from django.utils import timezone
from funkwhale_api.federation import utils as federation_utils


class LocalFromFidQuerySet:
    def local(self, include=True):
        host = settings.FEDERATION_HOSTNAME
        query = models.Q(fid__startswith="http://{}/".format(host)) | models.Q(
            fid__startswith="https://{}/".format(host)
        )
        if include:
            return self.filter(query)
        else:
            return self.filter(~query)


class MutationQuerySet(models.QuerySet):
    def get_for_target(self, target):
        content_type = ContentType.objects.get_for_model(target)
        return self.filter(target_content_type=content_type, target_id=target.pk)


class Mutation(models.Model):
    fid = models.URLField(unique=True, max_length=500, db_index=True)
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    created_by = models.ForeignKey(
        "federation.Actor",
        related_name="created_mutations",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    approved_by = models.ForeignKey(
        "federation.Actor",
        related_name="approved_mutations",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )

    type = models.CharField(max_length=100, db_index=True)
    # None = no choice, True = approved, False = refused
    is_approved = models.NullBooleanField(default=None)

    # None = not applied, True = applied, False = failed
    is_applied = models.NullBooleanField(default=None)
    creation_date = models.DateTimeField(default=timezone.now, db_index=True)
    applied_date = models.DateTimeField(null=True, blank=True, db_index=True)
    summary = models.TextField(max_length=2000, null=True, blank=True)

    payload = JSONField()
    previous_state = JSONField(null=True, default=None)

    target_id = models.IntegerField(null=True)
    target_content_type = models.ForeignKey(
        ContentType,
        null=True,
        on_delete=models.CASCADE,
        related_name="targeting_mutations",
    )
    target = GenericForeignKey("target_content_type", "target_id")

    objects = MutationQuerySet.as_manager()

    def get_federation_id(self):
        if self.fid:
            return self.fid

        return federation_utils.full_url(
            reverse("federation:edits-detail", kwargs={"uuid": self.uuid})
        )

    def save(self, **kwargs):
        if not self.pk and not self.fid:
            self.fid = self.get_federation_id()

        return super().save(**kwargs)

    @transaction.atomic
    def apply(self):
        from . import mutations

        if self.is_applied:
            raise ValueError("Mutation was already applied")

        previous_state = mutations.registry.apply(
            type=self.type, obj=self.target, payload=self.payload
        )
        self.previous_state = previous_state
        self.is_applied = True
        self.applied_date = timezone.now()
        self.save(update_fields=["is_applied", "applied_date", "previous_state"])
        return previous_state

           

api/funkwhale_api/favorites/models.py ¶

            from django.db import models
from django.utils import timezone
from funkwhale_api.music.models import Track


class TrackFavorite(models.Model):
    creation_date = models.DateTimeField(default=timezone.now)
    user = models.ForeignKey(
        "users.User", related_name="track_favorites", on_delete=models.CASCADE
    )
    track = models.ForeignKey(
        Track, related_name="track_favorites", on_delete=models.CASCADE
    )

    class Meta:
        unique_together = ("track", "user")
        ordering = ("-creation_date",)

    @classmethod
    def add(cls, track, user):
        favorite, created = cls.objects.get_or_create(user=user, track=track)
        return favorite

    def get_activity_url(self):
        return "{}/favorites/tracks/{}".format(self.user.get_activity_url(), self.pk)

           

api/funkwhale_api/federation/models.py ¶

            import tempfile
import uuid

from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import JSONField
from django.core.exceptions import ObjectDoesNotExist
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models
from django.urls import reverse
from django.utils import timezone
from funkwhale_api.common import session
from funkwhale_api.common import utils as common_utils
from funkwhale_api.common import validators as common_validators
from funkwhale_api.music import utils as music_utils

from . import utils as federation_utils

TYPE_CHOICES = [
    ("Person", "Person"),
    ("Application", "Application"),
    ("Group", "Group"),
    ("Organization", "Organization"),
    ("Service", "Service"),
]


def empty_dict():
    return {}


def get_shared_inbox_url():
    return federation_utils.full_url(reverse("federation:shared-inbox"))


class FederationMixin(models.Model):
    # federation id/url
    fid = models.URLField(unique=True, max_length=500, db_index=True)
    url = models.URLField(max_length=500, null=True, blank=True)

    class Meta:
        abstract = True


class ActorQuerySet(models.QuerySet):
    def local(self, include=True):
        if include:
            return self.filter(domain__name=settings.FEDERATION_HOSTNAME)
        return self.exclude(domain__name=settings.FEDERATION_HOSTNAME)

    def with_current_usage(self):
        qs = self
        for s in ["pending", "skipped", "errored", "finished"]:
            qs = qs.annotate(
                **{
                    "_usage_{}".format(s): models.Sum(
                        "libraries__uploads__size",
                        filter=models.Q(libraries__uploads__import_status=s),
                    )
                }
            )

        return qs

    def with_uploads_count(self):
        return self.annotate(
            uploads_count=models.Count("libraries__uploads", distinct=True)
        )


class DomainQuerySet(models.QuerySet):
    def external(self):
        return self.exclude(pk=settings.FEDERATION_HOSTNAME)

    def with_actors_count(self):
        return self.annotate(actors_count=models.Count("actors", distinct=True))

    def with_outbox_activities_count(self):
        return self.annotate(
            outbox_activities_count=models.Count(
                "actors__outbox_activities", distinct=True
            )
        )


class Domain(models.Model):
    name = models.CharField(
        primary_key=True,
        max_length=255,
        validators=[common_validators.DomainValidator()],
    )
    creation_date = models.DateTimeField(default=timezone.now)
    nodeinfo_fetch_date = models.DateTimeField(default=None, null=True, blank=True)
    nodeinfo = JSONField(default=empty_dict, max_length=50000, blank=True)
    service_actor = models.ForeignKey(
        "Actor",
        related_name="managed_domains",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    objects = DomainQuerySet.as_manager()

    def __str__(self):
        return self.name

    def save(self, **kwargs):
        lowercase_fields = ["name"]
        for field in lowercase_fields:
            v = getattr(self, field, None)
            if v:
                setattr(self, field, v.lower())

        super().save(**kwargs)

    def get_stats(self):
        from funkwhale_api.music import models as music_models

        data = Domain.objects.filter(pk=self.pk).aggregate(
            actors=models.Count("actors", distinct=True),
            outbox_activities=models.Count("actors__outbox_activities", distinct=True),
            libraries=models.Count("actors__libraries", distinct=True),
            received_library_follows=models.Count(
                "actors__libraries__received_follows", distinct=True
            ),
            emitted_library_follows=models.Count(
                "actors__library_follows", distinct=True
            ),
        )
        data["artists"] = music_models.Artist.objects.filter(
            from_activity__actor__domain_id=self.pk
        ).count()
        data["albums"] = music_models.Album.objects.filter(
            from_activity__actor__domain_id=self.pk
        ).count()
        data["tracks"] = music_models.Track.objects.filter(
            from_activity__actor__domain_id=self.pk
        ).count()

        uploads = music_models.Upload.objects.filter(library__actor__domain_id=self.pk)
        data["uploads"] = uploads.count()
        data["media_total_size"] = uploads.aggregate(v=models.Sum("size"))["v"] or 0
        data["media_downloaded_size"] = (
            uploads.with_file().aggregate(v=models.Sum("size"))["v"] or 0
        )
        return data


class Actor(models.Model):
    ap_type = "Actor"

    fid = models.URLField(unique=True, max_length=500, db_index=True)
    url = models.URLField(max_length=500, null=True, blank=True)
    outbox_url = models.URLField(max_length=500)
    inbox_url = models.URLField(max_length=500)
    following_url = models.URLField(max_length=500, null=True, blank=True)
    followers_url = models.URLField(max_length=500, null=True, blank=True)
    shared_inbox_url = models.URLField(max_length=500, null=True, blank=True)
    type = models.CharField(choices=TYPE_CHOICES, default="Person", max_length=25)
    name = models.CharField(max_length=200, null=True, blank=True)
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE, related_name="actors")
    summary = models.CharField(max_length=500, null=True, blank=True)
    preferred_username = models.CharField(max_length=200, null=True, blank=True)
    public_key = models.TextField(max_length=5000, null=True, blank=True)
    private_key = models.TextField(max_length=5000, null=True, blank=True)
    creation_date = models.DateTimeField(default=timezone.now)
    last_fetch_date = models.DateTimeField(default=timezone.now)
    manually_approves_followers = models.NullBooleanField(default=None)
    followers = models.ManyToManyField(
        to="self",
        symmetrical=False,
        through="Follow",
        through_fields=("target", "actor"),
        related_name="following",
    )

    objects = ActorQuerySet.as_manager()

    class Meta:
        unique_together = ["domain", "preferred_username"]

    @property
    def webfinger_subject(self):
        return "{}@{}".format(self.preferred_username, settings.FEDERATION_HOSTNAME)

    @property
    def private_key_id(self):
        return "{}#main-key".format(self.fid)

    @property
    def full_username(self):
        return "{}@{}".format(self.preferred_username, self.domain_id)

    def __str__(self):
        return "{}@{}".format(self.preferred_username, self.domain_id)

    @property
    def is_local(self):
        return self.domain_id == settings.FEDERATION_HOSTNAME

    def get_approved_followers(self):
        follows = self.received_follows.filter(approved=True)
        return self.followers.filter(pk__in=follows.values_list("actor", flat=True))

    def should_autoapprove_follow(self, actor):
        return False

    def get_user(self):
        try:
            return self.user
        except ObjectDoesNotExist:
            return None

    def get_current_usage(self):
        actor = self.__class__.objects.filter(pk=self.pk).with_current_usage().get()
        data = {}
        for s in ["pending", "skipped", "errored", "finished"]:
            data[s] = getattr(actor, "_usage_{}".format(s)) or 0

        data["total"] = sum(data.values())
        return data

    def get_stats(self):
        from funkwhale_api.music import models as music_models

        data = Actor.objects.filter(pk=self.pk).aggregate(
            outbox_activities=models.Count("outbox_activities", distinct=True),
            libraries=models.Count("libraries", distinct=True),
            received_library_follows=models.Count(
                "libraries__received_follows", distinct=True
            ),
            emitted_library_follows=models.Count("library_follows", distinct=True),
        )
        data["artists"] = music_models.Artist.objects.filter(
            from_activity__actor=self.pk
        ).count()
        data["albums"] = music_models.Album.objects.filter(
            from_activity__actor=self.pk
        ).count()
        data["tracks"] = music_models.Track.objects.filter(
            from_activity__actor=self.pk
        ).count()

        uploads = music_models.Upload.objects.filter(library__actor=self.pk)
        data["uploads"] = uploads.count()
        data["media_total_size"] = uploads.aggregate(v=models.Sum("size"))["v"] or 0
        data["media_downloaded_size"] = (
            uploads.with_file().aggregate(v=models.Sum("size"))["v"] or 0
        )
        return data

    @property
    def keys(self):
        return self.private_key, self.public_key

    @keys.setter
    def keys(self, v):
        self.private_key = v[0].decode("utf-8")
        self.public_key = v[1].decode("utf-8")


class InboxItem(models.Model):
    """
    Store activities binding to local actors, with read/unread status.
    """

    actor = models.ForeignKey(
        Actor, related_name="inbox_items", on_delete=models.CASCADE
    )
    activity = models.ForeignKey(
        "Activity", related_name="inbox_items", on_delete=models.CASCADE
    )
    type = models.CharField(max_length=10, choices=[("to", "to"), ("cc", "cc")])
    is_read = models.BooleanField(default=False)


class Delivery(models.Model):
    """
    Store deliveries attempt to remote inboxes
    """

    is_delivered = models.BooleanField(default=False)
    last_attempt_date = models.DateTimeField(null=True, blank=True)
    attempts = models.PositiveIntegerField(default=0)
    inbox_url = models.URLField(max_length=500)

    activity = models.ForeignKey(
        "Activity", related_name="deliveries", on_delete=models.CASCADE
    )


class Activity(models.Model):
    actor = models.ForeignKey(
        Actor, related_name="outbox_activities", on_delete=models.CASCADE
    )
    recipients = models.ManyToManyField(
        Actor, related_name="inbox_activities", through=InboxItem
    )
    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
    fid = models.URLField(unique=True, max_length=500, null=True, blank=True)
    url = models.URLField(max_length=500, null=True, blank=True)
    payload = JSONField(default=empty_dict, max_length=50000, encoder=DjangoJSONEncoder)
    creation_date = models.DateTimeField(default=timezone.now, db_index=True)
    type = models.CharField(db_index=True, null=True, max_length=100)

    # generic relations
    object_id = models.IntegerField(null=True)
    object_content_type = models.ForeignKey(
        ContentType,
        null=True,
        on_delete=models.SET_NULL,
        related_name="objecting_activities",
    )
    object = GenericForeignKey("object_content_type", "object_id")
    target_id = models.IntegerField(null=True)
    target_content_type = models.ForeignKey(
        ContentType,
        null=True,
        on_delete=models.SET_NULL,
        related_name="targeting_activities",
    )
    target = GenericForeignKey("target_content_type", "target_id")
    related_object_id = models.IntegerField(null=True)
    related_object_content_type = models.ForeignKey(
        ContentType,
        null=True,
        on_delete=models.SET_NULL,
        related_name="related_objecting_activities",
    )
    related_object = GenericForeignKey(
        "related_object_content_type", "related_object_id"
    )


class AbstractFollow(models.Model):
    ap_type = "Follow"
    fid = models.URLField(unique=True, max_length=500, null=True, blank=True)
    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
    creation_date = models.DateTimeField(default=timezone.now)
    modification_date = models.DateTimeField(auto_now=True)
    approved = models.NullBooleanField(default=None)

    class Meta:
        abstract = True

    def get_federation_id(self):
        return federation_utils.full_url(
            "{}#follows/{}".format(self.actor.fid, self.uuid)
        )


class Follow(AbstractFollow):
    actor = models.ForeignKey(
        Actor, related_name="emitted_follows", on_delete=models.CASCADE
    )
    target = models.ForeignKey(
        Actor, related_name="received_follows", on_delete=models.CASCADE
    )

    class Meta:
        unique_together = ["actor", "target"]


class LibraryFollow(AbstractFollow):
    actor = models.ForeignKey(
        Actor, related_name="library_follows", on_delete=models.CASCADE
    )
    target = models.ForeignKey(
        "music.Library", related_name="received_follows", on_delete=models.CASCADE
    )

    class Meta:
        unique_together = ["actor", "target"]


class Library(models.Model):
    creation_date = models.DateTimeField(default=timezone.now)
    modification_date = models.DateTimeField(auto_now=True)
    fetched_date = models.DateTimeField(null=True, blank=True)
    actor = models.OneToOneField(
        Actor, on_delete=models.CASCADE, related_name="library"
    )
    uuid = models.UUIDField(default=uuid.uuid4)
    url = models.URLField(max_length=500)

    # use this flag to disable federation with a library
    federation_enabled = models.BooleanField()
    # should we mirror files locally or hotlink them?
    download_files = models.BooleanField()
    # should we automatically import new files from this library?
    autoimport = models.BooleanField()
    tracks_count = models.PositiveIntegerField(null=True, blank=True)
    follow = models.OneToOneField(
        Follow, related_name="library", null=True, blank=True, on_delete=models.SET_NULL
    )


get_file_path = common_utils.ChunkedPath("federation_cache")


class LibraryTrack(models.Model):
    url = models.URLField(unique=True, max_length=500)
    audio_url = models.URLField(max_length=500)
    audio_mimetype = models.CharField(max_length=200)
    audio_file = models.FileField(upload_to=get_file_path, null=True, blank=True)

    creation_date = models.DateTimeField(default=timezone.now)
    modification_date = models.DateTimeField(auto_now=True)
    fetched_date = models.DateTimeField(null=True, blank=True)
    published_date = models.DateTimeField(null=True, blank=True)
    library = models.ForeignKey(
        Library, related_name="tracks", on_delete=models.CASCADE
    )
    artist_name = models.CharField(max_length=500)
    album_title = models.CharField(max_length=500)
    title = models.CharField(max_length=500)
    metadata = JSONField(
        default=empty_dict, max_length=10000, encoder=DjangoJSONEncoder
    )

    @property
    def mbid(self):
        try:
            return self.metadata["recording"]["musicbrainz_id"]
        except KeyError:
            pass

    def download_audio(self):
        from . import actors

        auth = actors.SYSTEM_ACTORS["library"].get_request_auth()
        remote_response = session.get_session().get(
            self.audio_url,
            auth=auth,
            stream=True,
            timeout=20,
            verify=settings.EXTERNAL_REQUESTS_VERIFY_SSL,
            headers={"Content-Type": "application/activity+json"},
        )
        with remote_response as r:
            remote_response.raise_for_status()
            extension = music_utils.get_ext_from_type(self.audio_mimetype)
            title = " - ".join([self.title, self.album_title, self.artist_name])
            filename = "{}.{}".format(title, extension)
            tmp_file = tempfile.TemporaryFile()
            for chunk in r.iter_content(chunk_size=512):
                tmp_file.write(chunk)
            self.audio_file.save(filename, tmp_file)

    def get_metadata(self, key):
        return self.metadata.get(key)

           

api/funkwhale_api/history/models.py ¶

            from django.db import models
from django.utils import timezone
from funkwhale_api.music.models import Track


class Listening(models.Model):
    creation_date = models.DateTimeField(default=timezone.now, null=True, blank=True)
    track = models.ForeignKey(
        Track, related_name="listenings", on_delete=models.CASCADE
    )
    user = models.ForeignKey(
        "users.User",
        related_name="listenings",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
    )
    session_key = models.CharField(max_length=100, null=True, blank=True)

    class Meta:
        ordering = ("-creation_date",)

    def get_activity_url(self):
        return "{}/listenings/tracks/{}".format(self.user.get_activity_url(), self.pk)

           

api/funkwhale_api/moderation/models.py ¶

            import urllib.parse
import uuid

from django.db import models
from django.utils import timezone


class InstancePolicyQuerySet(models.QuerySet):
    def active(self):
        return self.filter(is_active=True)

    def matching_url(self, *urls):
        if not urls:
            return self.none()
        query = None
        for url in urls:
            new_query = self.matching_url_query(url)
            if query:
                query = query | new_query
            else:
                query = new_query
        return self.filter(query)

    def matching_url_query(self, url):
        parsed = urllib.parse.urlparse(url)
        return models.Q(target_domain_id=parsed.hostname) | models.Q(
            target_actor__fid=url
        )


class InstancePolicy(models.Model):
    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
    actor = models.ForeignKey(
        "federation.Actor",
        related_name="created_instance_policies",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    target_domain = models.OneToOneField(
        "federation.Domain",
        related_name="instance_policy",
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )
    target_actor = models.OneToOneField(
        "federation.Actor",
        related_name="instance_policy",
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )
    creation_date = models.DateTimeField(default=timezone.now)

    is_active = models.BooleanField(default=True)
    # a summary explaining why the policy is in place
    summary = models.TextField(max_length=10000, null=True, blank=True)
    # either block everything (simpler, but less granularity)
    block_all = models.BooleanField(default=False)
    # or pick individual restrictions below
    # do not show in timelines/notifications, except for actual followers
    silence_activity = models.BooleanField(default=False)
    silence_notifications = models.BooleanField(default=False)
    # do not download any media from the target
    reject_media = models.BooleanField(default=False)

    objects = InstancePolicyQuerySet.as_manager()

    @property
    def target(self):
        if self.target_actor:
            return {"type": "actor", "obj": self.target_actor}
        if self.target_domain_id:
            return {"type": "domain", "obj": self.target_domain}


class UserFilter(models.Model):
    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
    creation_date = models.DateTimeField(default=timezone.now)
    target_artist = models.ForeignKey(
        "music.Artist", on_delete=models.CASCADE, related_name="user_filters"
    )
    user = models.ForeignKey(
        "users.User", on_delete=models.CASCADE, related_name="content_filters"
    )

    class Meta:
        unique_together = ("user", "target_artist")

    @property
    def target(self):
        if self.target_artist:
            return {"type": "artist", "obj": self.target_artist}

           

api/funkwhale_api/music/models.py ¶

            import datetime
import logging
import mimetypes
import os
import tempfile
import uuid

import markdown
import pendulum
import pydub
from django.conf import settings
from django.contrib.postgres.fields import JSONField
from django.core.files.base import ContentFile
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models
from django.db import transaction
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.urls import reverse
from django.utils import timezone
from funkwhale_api import musicbrainz
from funkwhale_api.common import fields
from funkwhale_api.common import models as common_models
from funkwhale_api.common import session
from funkwhale_api.common import utils as common_utils
from funkwhale_api.federation import models as federation_models
from funkwhale_api.federation import utils as federation_utils
from taggit.managers import TaggableManager
from versatileimagefield.fields import VersatileImageField
from versatileimagefield.image_warmer import VersatileImageFieldWarmer

from . import importers
from . import metadata
from . import utils

logger = logging.getLogger(__name__)


def empty_dict():
    return {}


class APIModelMixin(models.Model):
    fid = models.URLField(unique=True, max_length=500, db_index=True, null=True)
    mbid = models.UUIDField(unique=True, db_index=True, null=True, blank=True)
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    from_activity = models.ForeignKey(
        "federation.Activity", null=True, blank=True, on_delete=models.SET_NULL
    )
    api_includes = []
    creation_date = models.DateTimeField(default=timezone.now, db_index=True)
    import_hooks = []

    class Meta:
        abstract = True
        ordering = ["-creation_date"]

    @classmethod
    def get_or_create_from_api(cls, mbid):
        try:
            return cls.objects.get(mbid=mbid), False
        except cls.DoesNotExist:
            return cls.create_from_api(id=mbid), True

    def get_api_data(self):
        return self.__class__.api.get(id=self.mbid, includes=self.api_includes)[
            self.musicbrainz_model
        ]

    @classmethod
    def create_from_api(cls, **kwargs):
        if kwargs.get("id"):
            raw_data = cls.api.get(id=kwargs["id"], includes=cls.api_includes)[
                cls.musicbrainz_model
            ]
        else:
            raw_data = cls.api.search(**kwargs)[
                "{0}-list".format(cls.musicbrainz_model)
            ][0]
        cleaned_data = cls.clean_musicbrainz_data(raw_data)
        return importers.load(cls, cleaned_data, raw_data, cls.import_hooks)

    @classmethod
    def clean_musicbrainz_data(cls, data):
        cleaned_data = {}
        mapping = importers.Mapping(cls.musicbrainz_mapping)
        for key, value in data.items():
            try:
                cleaned_key, cleaned_value = mapping.from_musicbrainz(key, value)
                cleaned_data[cleaned_key] = cleaned_value
            except KeyError:
                pass
        return cleaned_data

    @property
    def musicbrainz_url(self):
        if self.mbid:
            return "https://musicbrainz.org/{}/{}".format(
                self.musicbrainz_model, self.mbid
            )

    def get_federation_id(self):
        if self.fid:
            return self.fid

        return federation_utils.full_url(
            reverse(
                "federation:music:{}-detail".format(self.federation_namespace),
                kwargs={"uuid": self.uuid},
            )
        )

    def save(self, **kwargs):
        if not self.pk and not self.fid:
            self.fid = self.get_federation_id()

        return super().save(**kwargs)


class License(models.Model):
    code = models.CharField(primary_key=True, max_length=100)
    url = models.URLField(max_length=500)

    # if true, license is a copyleft license, meaning that derivative
    # work must be shared under the same license
    copyleft = models.BooleanField()
    # if true, commercial use of the work is allowed
    commercial = models.BooleanField()
    # if true, attribution to the original author is required when reusing
    # the work
    attribution = models.BooleanField()
    # if true, derivative work are allowed
    derivative = models.BooleanField()
    # if true, redistribution of the wor is allowed
    redistribute = models.BooleanField()

    @property
    def conf(self):
        from . import licenses

        for row in licenses.LICENSES:
            if self.code == row["code"]:
                return row
        logger.warning("%s do not match any registered license", self.code)


class ArtistQuerySet(common_models.LocalFromFidQuerySet, models.QuerySet):
    def with_albums_count(self):
        return self.annotate(_albums_count=models.Count("albums"))

    def with_albums(self):
        return self.prefetch_related(
            models.Prefetch("albums", queryset=Album.objects.with_tracks_count())
        )

    def annotate_playable_by_actor(self, actor):
        tracks = (
            Upload.objects.playable_by(actor)
            .filter(track__artist=models.OuterRef("id"))
            .order_by("id")
            .values("id")[:1]
        )
        subquery = models.Subquery(tracks)
        return self.annotate(is_playable_by_actor=subquery)

    def playable_by(self, actor, include=True):
        tracks = Track.objects.playable_by(actor, include)
        matches = self.filter(tracks__in=tracks).values_list("pk")
        if include:
            return self.filter(pk__in=matches)
        else:
            return self.exclude(pk__in=matches)


class Artist(APIModelMixin):
    name = models.CharField(max_length=255)
    federation_namespace = "artists"
    musicbrainz_model = "artist"
    musicbrainz_mapping = {
        "mbid": {"musicbrainz_field_name": "id"},
        "name": {"musicbrainz_field_name": "name"},
    }
    api = musicbrainz.api.artists
    objects = ArtistQuerySet.as_manager()

    def __str__(self):
        return self.name

    @property
    def tags(self):
        t = []
        for album in self.albums.all():
            for tag in album.tags:
                t.append(tag)
        return set(t)

    @classmethod
    def get_or_create_from_name(cls, name, **kwargs):
        kwargs.update({"name": name})
        return cls.objects.get_or_create(name__iexact=name, defaults=kwargs)


def import_artist(v):
    a = Artist.get_or_create_from_api(mbid=v[0]["artist"]["id"])[0]
    return a


def parse_date(v):
    d = pendulum.parse(v).date()
    return d


def import_tracks(instance, cleaned_data, raw_data):
    for track_data in raw_data["medium-list"][0]["track-list"]:
        track_cleaned_data = Track.clean_musicbrainz_data(track_data["recording"])
        track_cleaned_data["album"] = instance
        track_cleaned_data["position"] = int(track_data["position"])
        importers.load(Track, track_cleaned_data, track_data, Track.import_hooks)


class AlbumQuerySet(common_models.LocalFromFidQuerySet, models.QuerySet):
    def with_tracks_count(self):
        return self.annotate(_tracks_count=models.Count("tracks"))

    def annotate_playable_by_actor(self, actor):
        tracks = (
            Upload.objects.playable_by(actor)
            .filter(track__album=models.OuterRef("id"))
            .order_by("id")
            .values("id")[:1]
        )
        subquery = models.Subquery(tracks)
        return self.annotate(is_playable_by_actor=subquery)

    def playable_by(self, actor, include=True):
        tracks = Track.objects.playable_by(actor, include)
        matches = self.filter(tracks__in=tracks).values_list("pk")
        if include:
            return self.filter(pk__in=matches)
        else:
            return self.exclude(pk__in=matches)

    def with_prefetched_tracks_and_playable_uploads(self, actor):
        tracks = Track.objects.with_playable_uploads(actor)
        return self.prefetch_related(models.Prefetch("tracks", queryset=tracks))


class Album(APIModelMixin):
    title = models.CharField(max_length=255)
    artist = models.ForeignKey(Artist, related_name="albums", on_delete=models.CASCADE)
    release_date = models.DateField(null=True, blank=True)
    release_group_id = models.UUIDField(null=True, blank=True)
    cover = VersatileImageField(
        upload_to="albums/covers/%Y/%m/%d", null=True, blank=True
    )
    TYPE_CHOICES = (("album", "Album"),)
    type = models.CharField(choices=TYPE_CHOICES, max_length=30, default="album")

    api_includes = ["artist-credits", "recordings", "media", "release-groups"]
    api = musicbrainz.api.releases
    federation_namespace = "albums"
    musicbrainz_model = "release"
    musicbrainz_mapping = {
        "mbid": {"musicbrainz_field_name": "id"},
        "position": {
            "musicbrainz_field_name": "release-list",
            "converter": lambda v: int(v[0]["medium-list"][0]["position"]),
        },
        "release_group_id": {
            "musicbrainz_field_name": "release-group",
            "converter": lambda v: v["id"],
        },
        "title": {"musicbrainz_field_name": "title"},
        "release_date": {"musicbrainz_field_name": "date", "converter": parse_date},
        "type": {"musicbrainz_field_name": "type", "converter": lambda v: v.lower()},
        "artist": {
            "musicbrainz_field_name": "artist-credit",
            "converter": import_artist,
        },
    }
    objects = AlbumQuerySet.as_manager()

    def get_image(self, data=None):
        if data:
            extensions = {"image/jpeg": "jpg", "image/png": "png", "image/gif": "gif"}
            extension = extensions.get(data["mimetype"], "jpg")
            if data.get("content"):
                # we have to cover itself
                f = ContentFile(data["content"])
            elif data.get("url"):
                # we can fetch from a url
                try:
                    response = session.get_session().get(
                        data.get("url"),
                        timeout=3,
                        verify=settings.EXTERNAL_REQUESTS_VERIFY_SSL,
                    )
                    response.raise_for_status()
                except Exception as e:
                    logger.warn(
                        "Cannot download cover at url %s: %s", data.get("url"), e
                    )
                    return
                else:
                    f = ContentFile(response.content)
            self.cover.save("{}.{}".format(self.uuid, extension), f, save=False)
            self.save(update_fields=["cover"])
            return self.cover.file
        if self.mbid:
            image_data = musicbrainz.api.images.get_front(str(self.mbid))
            f = ContentFile(image_data)
            self.cover.save("{0}.jpg".format(self.mbid), f, save=False)
            self.save(update_fields=["cover"])
        return self.cover.file

    def __str__(self):
        return self.title

    @property
    def tags(self):
        t = []
        for track in self.tracks.all():
            for tag in track.tags.all():
                t.append(tag)
        return set(t)

    @classmethod
    def get_or_create_from_title(cls, title, **kwargs):
        kwargs.update({"title": title})
        return cls.objects.get_or_create(title__iexact=title, defaults=kwargs)


def import_tags(instance, cleaned_data, raw_data):
    MINIMUM_COUNT = 2
    tags_to_add = []
    for tag_data in raw_data.get("tag-list", []):
        try:
            if int(tag_data["count"]) < MINIMUM_COUNT:
                continue
        except ValueError:
            continue
        tags_to_add.append(tag_data["name"])
    instance.tags.add(*tags_to_add)


def import_album(v):
    a = Album.get_or_create_from_api(mbid=v[0]["id"])[0]
    return a


def link_recordings(instance, cleaned_data, raw_data):
    tracks = [r["target"] for r in raw_data["recording-relation-list"]]
    Track.objects.filter(mbid__in=tracks).update(work=instance)


def import_lyrics(instance, cleaned_data, raw_data):
    try:
        url = [
            url_data
            for url_data in raw_data["url-relation-list"]
            if url_data["type"] == "lyrics"
        ][0]["target"]
    except (IndexError, KeyError):
        return
    l, _ = Lyrics.objects.get_or_create(work=instance, url=url)

    return l


class Work(APIModelMixin):
    language = models.CharField(max_length=20)
    nature = models.CharField(max_length=50)
    title = models.CharField(max_length=255)

    api = musicbrainz.api.works
    api_includes = ["url-rels", "recording-rels"]
    musicbrainz_model = "work"
    federation_namespace = "works"

    musicbrainz_mapping = {
        "mbid": {"musicbrainz_field_name": "id"},
        "title": {"musicbrainz_field_name": "title"},
        "language": {"musicbrainz_field_name": "language"},
        "nature": {"musicbrainz_field_name": "type", "converter": lambda v: v.lower()},
    }
    import_hooks = [import_lyrics, link_recordings]

    def fetch_lyrics(self):
        lyric = self.lyrics.first()
        if lyric:
            return lyric
        data = self.api.get(self.mbid, includes=["url-rels"])["work"]
        lyric = import_lyrics(self, {}, data)

        return lyric

    def get_federation_id(self):
        if self.fid:
            return self.fid

        return None


class Lyrics(models.Model):
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    work = models.ForeignKey(
        Work, related_name="lyrics", null=True, blank=True, on_delete=models.CASCADE
    )
    url = models.URLField(unique=True)
    content = models.TextField(null=True, blank=True)

    @property
    def content_rendered(self):
        return markdown.markdown(
            self.content,
            safe_mode=True,
            enable_attributes=False,
            extensions=["markdown.extensions.nl2br"],
        )


class TrackQuerySet(common_models.LocalFromFidQuerySet, models.QuerySet):
    def for_nested_serialization(self):
        return self.select_related().select_related("album__artist", "artist")

    def annotate_playable_by_actor(self, actor):
        files = (
            Upload.objects.playable_by(actor)
            .filter(track=models.OuterRef("id"))
            .order_by("id")
            .values("id")[:1]
        )
        subquery = models.Subquery(files)
        return self.annotate(is_playable_by_actor=subquery)

    def playable_by(self, actor, include=True):
        files = Upload.objects.playable_by(actor, include)
        matches = self.filter(uploads__in=files).values_list("pk")
        if include:
            return self.filter(pk__in=matches)
        else:
            return self.exclude(pk__in=matches)

    def with_playable_uploads(self, actor):
        uploads = Upload.objects.playable_by(actor).select_related("track")
        return self.prefetch_related(
            models.Prefetch("uploads", queryset=uploads, to_attr="playable_uploads")
        )

    def order_for_album(self):
        """
        Order by disc number then position
        """
        return self.order_by("disc_number", "position", "title")


def get_artist(release_list):
    return Artist.get_or_create_from_api(
        mbid=release_list[0]["artist-credits"][0]["artists"]["id"]
    )[0]


class Track(APIModelMixin):
    title = models.CharField(max_length=255)
    artist = models.ForeignKey(Artist, related_name="tracks", on_delete=models.CASCADE)
    disc_number = models.PositiveIntegerField(null=True, blank=True)
    position = models.PositiveIntegerField(null=True, blank=True)
    album = models.ForeignKey(
        Album, related_name="tracks", null=True, blank=True, on_delete=models.CASCADE
    )
    work = models.ForeignKey(
        Work, related_name="tracks", null=True, blank=True, on_delete=models.CASCADE
    )
    license = models.ForeignKey(
        License,
        null=True,
        blank=True,
        on_delete=models.DO_NOTHING,
        related_name="tracks",
    )
    copyright = models.CharField(max_length=500, null=True, blank=True)
    federation_namespace = "tracks"
    musicbrainz_model = "recording"
    api = musicbrainz.api.recordings
    api_includes = ["artist-credits", "releases", "media", "tags", "work-rels"]
    musicbrainz_mapping = {
        "mbid": {"musicbrainz_field_name": "id"},
        "title": {"musicbrainz_field_name": "title"},
        "artist": {
            "musicbrainz_field_name": "artist-credit",
            "converter": lambda v: Artist.get_or_create_from_api(
                mbid=v[0]["artist"]["id"]
            )[0],
        },
        "album": {"musicbrainz_field_name": "release-list", "converter": import_album},
    }
    import_hooks = [import_tags]
    objects = TrackQuerySet.as_manager()
    tags = TaggableManager(blank=True)

    class Meta:
        ordering = ["album", "disc_number", "position"]

    def __str__(self):
        return self.title

    def save(self, **kwargs):
        try:
            self.artist
        except Artist.DoesNotExist:
            self.artist = self.album.artist
        super().save(**kwargs)

    def get_work(self):
        if self.work:
            return self.work
        data = self.api.get(self.mbid, includes=["work-rels"])
        try:
            work_data = data["recording"]["work-relation-list"][0]["work"]
        except (IndexError, KeyError):
            return
        work, _ = Work.get_or_create_from_api(mbid=work_data["id"])
        return work

    def get_lyrics_url(self):
        return reverse("api:v1:tracks-lyrics", kwargs={"pk": self.pk})

    @property
    def full_name(self):
        try:
            return "{} - {} - {}".format(self.artist.name, self.album.title, self.title)
        except AttributeError:
            return "{} - {}".format(self.artist.name, self.title)

    def get_activity_url(self):
        if self.mbid:
            return "https://musicbrainz.org/recording/{}".format(self.mbid)
        return settings.FUNKWHALE_URL + "/tracks/{}".format(self.pk)

    @classmethod
    def get_or_create_from_title(cls, title, **kwargs):
        kwargs.update({"title": title})
        return cls.objects.get_or_create(title__iexact=title, defaults=kwargs)

    @classmethod
    def get_or_create_from_release(cls, release_mbid, mbid):
        release_mbid = str(release_mbid)
        mbid = str(mbid)
        try:
            return cls.objects.get(mbid=mbid), False
        except cls.DoesNotExist:
            pass

        album = Album.get_or_create_from_api(release_mbid)[0]
        data = musicbrainz.client.api.releases.get(
            str(album.mbid), includes=Album.api_includes
        )
        tracks = [t for m in data["release"]["medium-list"] for t in m["track-list"]]
        track_data = None
        for track in tracks:
            if track["recording"]["id"] == str(mbid):
                track_data = track
                break
        if not track_data:
            raise ValueError("No track found matching this ID")

        track_artist_mbid = None
        for ac in track_data["recording"]["artist-credit"]:
            try:
                ac_mbid = ac["artist"]["id"]
            except TypeError:
                # it's probably a string, like "feat."
                continue

            if ac_mbid == str(album.artist.mbid):
                continue

            track_artist_mbid = ac_mbid
            break
        track_artist_mbid = track_artist_mbid or album.artist.mbid
        if track_artist_mbid == str(album.artist.mbid):
            track_artist = album.artist
        else:
            track_artist = Artist.get_or_create_from_api(track_artist_mbid)[0]
        return cls.objects.update_or_create(
            mbid=mbid,
            defaults={
                "position": int(track["position"]),
                "title": track["recording"]["title"],
                "album": album,
                "artist": track_artist,
            },
        )

    @property
    def listen_url(self):
        return reverse("api:v1:listen-detail", kwargs={"uuid": self.uuid})

    @property
    def local_license(self):
        """
        Since license primary keys are strings, and we can get the data
        from our hardcoded licenses.LICENSES list, there is no need
        for extra SQL joins / queries.
        """
        from . import licenses

        return licenses.LICENSES_BY_ID.get(self.license_id)


class UploadQuerySet(models.QuerySet):
    def playable_by(self, actor, include=True):
        libraries = Library.objects.viewable_by(actor)

        if include:
            return self.filter(library__in=libraries, import_status="finished")
        return self.exclude(library__in=libraries, import_status="finished")

    def local(self, include=True):
        return self.exclude(library__actor__user__isnull=include)

    def for_federation(self):
        return self.filter(import_status="finished", mimetype__startswith="audio/")

    def with_file(self):
        return self.exclude(audio_file=None).exclude(audio_file="")


TRACK_FILE_IMPORT_STATUS_CHOICES = (
    ("pending", "Pending"),
    ("finished", "Finished"),
    ("errored", "Errored"),
    ("skipped", "Skipped"),
)


def get_file_path(instance, filename):
    if isinstance(instance, UploadVersion):
        return common_utils.ChunkedPath("transcoded")(instance, filename)

    if instance.library.actor.get_user():
        return common_utils.ChunkedPath("tracks")(instance, filename)
    else:
        # we cache remote tracks in a different directory
        return common_utils.ChunkedPath("federation_cache/tracks")(instance, filename)


def get_import_reference():
    return str(uuid.uuid4())


class Upload(models.Model):
    fid = models.URLField(unique=True, max_length=500, null=True, blank=True)
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    track = models.ForeignKey(
        Track, related_name="uploads", on_delete=models.CASCADE, null=True, blank=True
    )
    audio_file = models.FileField(upload_to=get_file_path, max_length=255)
    source = models.CharField(
        # URL validators are not flexible enough for our file:// and upload:// schemes
        null=True,
        blank=True,
        max_length=500,
    )
    creation_date = models.DateTimeField(default=timezone.now, db_index=True)
    modification_date = models.DateTimeField(default=timezone.now, null=True)
    accessed_date = models.DateTimeField(null=True, blank=True)
    duration = models.IntegerField(null=True, blank=True)
    size = models.IntegerField(null=True, blank=True)
    bitrate = models.IntegerField(null=True, blank=True)
    acoustid_track_id = models.UUIDField(null=True, blank=True)
    mimetype = models.CharField(null=True, blank=True, max_length=200)
    library = models.ForeignKey(
        "library",
        null=True,
        blank=True,
        related_name="uploads",
        on_delete=models.CASCADE,
    )

    # metadata from federation
    metadata = JSONField(
        default=empty_dict, max_length=50000, encoder=DjangoJSONEncoder
    )
    import_date = models.DateTimeField(null=True, blank=True)
    # optionnal metadata provided during import
    import_metadata = JSONField(
        default=empty_dict, max_length=50000, encoder=DjangoJSONEncoder
    )
    # status / error details for the import
    import_status = models.CharField(
        default="pending", choices=TRACK_FILE_IMPORT_STATUS_CHOICES, max_length=25
    )
    # a short reference provided by the client to group multiple files
    # in the same import
    import_reference = models.CharField(max_length=50, default=get_import_reference)

    # optionnal metadata about import results (error messages, etc.)
    import_details = JSONField(
        default=empty_dict, max_length=50000, encoder=DjangoJSONEncoder
    )
    from_activity = models.ForeignKey(
        "federation.Activity", null=True, on_delete=models.SET_NULL
    )

    objects = UploadQuerySet.as_manager()

    def download_audio_from_remote(self, user):
        from funkwhale_api.common import session
        from funkwhale_api.federation import signing

        if user.is_authenticated and user.actor:
            auth = signing.get_auth(user.actor.private_key, user.actor.private_key_id)
        else:
            auth = None

        remote_response = session.get_session().get(
            self.source,
            auth=auth,
            stream=True,
            timeout=20,
            headers={"Content-Type": "application/octet-stream"},
            verify=settings.EXTERNAL_REQUESTS_VERIFY_SSL,
        )
        with remote_response as r:
            remote_response.raise_for_status()
            extension = utils.get_ext_from_type(self.mimetype)
            title = " - ".join(
                [self.track.title, self.track.album.title, self.track.artist.name]
            )
            filename = "{}.{}".format(title, extension)
            tmp_file = tempfile.TemporaryFile()
            for chunk in r.iter_content(chunk_size=512):
                tmp_file.write(chunk)
            self.audio_file.save(filename, tmp_file, save=False)
            self.save(update_fields=["audio_file"])

    def get_federation_id(self):
        if self.fid:
            return self.fid

        return federation_utils.full_url(
            reverse("federation:music:uploads-detail", kwargs={"uuid": self.uuid})
        )

    @property
    def filename(self):
        return "{}.{}".format(self.track.full_name, self.extension)

    @property
    def extension(self):
        try:
            return utils.MIMETYPE_TO_EXTENSION[self.mimetype]
        except KeyError:
            pass
        if self.audio_file:
            return os.path.splitext(self.audio_file.name)[-1].replace(".", "", 1)
        if self.in_place_path:
            return os.path.splitext(self.in_place_path)[-1].replace(".", "", 1)

    def get_file_size(self):
        if self.audio_file:
            return self.audio_file.size

        if self.source.startswith("file://"):
            return os.path.getsize(self.source.replace("file://", "", 1))

    def get_audio_file(self):
        if self.audio_file:
            return self.audio_file.open()
        if self.source.startswith("file://"):
            return open(self.source.replace("file://", "", 1), "rb")

    def get_audio_data(self):
        audio_file = self.get_audio_file()
        if not audio_file:
            return
        audio_data = utils.get_audio_file_data(audio_file)
        if not audio_data:
            return
        return {
            "duration": int(audio_data["length"]),
            "bitrate": audio_data["bitrate"],
            "size": self.get_file_size(),
        }

    def get_audio_segment(self):
        input = self.get_audio_file()
        if not input:
            return

        input_format = utils.MIMETYPE_TO_EXTENSION[self.mimetype]
        audio = pydub.AudioSegment.from_file(input, format=input_format)
        return audio

    def save(self, **kwargs):
        if not self.mimetype:
            if self.audio_file:
                self.mimetype = utils.guess_mimetype(self.audio_file)
            elif self.source and self.source.startswith("file://"):
                self.mimetype = mimetypes.guess_type(self.source)[0]
        if not self.size and self.audio_file:
            self.size = self.audio_file.size
        if not self.pk and not self.fid and self.library.actor.get_user():
            self.fid = self.get_federation_id()
        return super().save(**kwargs)

    def get_metadata(self):
        audio_file = self.get_audio_file()
        if not audio_file:
            return
        return metadata.Metadata(audio_file)

    @property
    def listen_url(self):
        return self.track.listen_url + "?upload={}".format(self.uuid)

    def get_transcoded_version(self, format):
        mimetype = utils.EXTENSION_TO_MIMETYPE[format]
        existing_versions = list(self.versions.filter(mimetype=mimetype))
        if existing_versions:
            # we found an existing version, no need to transcode again
            return existing_versions[0]

        return self.create_transcoded_version(mimetype, format)

    @transaction.atomic
    def create_transcoded_version(self, mimetype, format):
        # we create the version with an empty file, then
        # we'll write to it
        f = ContentFile(b"")
        version = self.versions.create(
            mimetype=mimetype, bitrate=self.bitrate or 128000, size=0
        )
        # we keep the same name, but we update the extension
        new_name = os.path.splitext(os.path.basename(self.audio_file.name))[
            0
        ] + ".{}".format(format)
        version.audio_file.save(new_name, f)
        utils.transcode_audio(
            audio=self.get_audio_segment(),
            output=version.audio_file,
            output_format=utils.MIMETYPE_TO_EXTENSION[mimetype],
        )
        version.size = version.audio_file.size
        version.save(update_fields=["size"])

        return version

    @property
    def in_place_path(self):
        if not self.source or not self.source.startswith("file://"):
            return
        return self.source.lstrip("file://")


MIMETYPE_CHOICES = [(mt, ext) for ext, mt in utils.AUDIO_EXTENSIONS_AND_MIMETYPE]


class UploadVersion(models.Model):
    upload = models.ForeignKey(
        Upload, related_name="versions", on_delete=models.CASCADE
    )
    mimetype = models.CharField(max_length=50, choices=MIMETYPE_CHOICES)
    creation_date = models.DateTimeField(default=timezone.now)
    accessed_date = models.DateTimeField(null=True, blank=True)
    audio_file = models.FileField(upload_to=get_file_path, max_length=255)
    bitrate = models.PositiveIntegerField()
    size = models.IntegerField()

    class Meta:
        unique_together = ("upload", "mimetype", "bitrate")

    @property
    def filename(self):
        return self.upload.filename


IMPORT_STATUS_CHOICES = (
    ("pending", "Pending"),
    ("finished", "Finished"),
    ("errored", "Errored"),
    ("skipped", "Skipped"),
)


class ImportBatch(models.Model):
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    IMPORT_BATCH_SOURCES = [
        ("api", "api"),
        ("shell", "shell"),
        ("federation", "federation"),
    ]
    source = models.CharField(
        max_length=30, default="api", choices=IMPORT_BATCH_SOURCES
    )
    creation_date = models.DateTimeField(default=timezone.now)
    submitted_by = models.ForeignKey(
        "users.User",
        related_name="imports",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
    )
    status = models.CharField(
        choices=IMPORT_STATUS_CHOICES, default="pending", max_length=30
    )
    import_request = models.ForeignKey(
        "requests.ImportRequest",
        related_name="import_batches",
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    library = models.ForeignKey(
        "Library",
        related_name="import_batches",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
    )

    class Meta:
        ordering = ["-creation_date"]

    def __str__(self):
        return str(self.pk)

    def update_status(self):
        old_status = self.status
        self.status = utils.compute_status(self.jobs.all())
        if self.status == old_status:
            return
        self.save(update_fields=["status"])
        if self.status != old_status and self.status == "finished":
            from . import tasks

            tasks.import_batch_notify_followers.delay(import_batch_id=self.pk)

    def get_federation_id(self):
        return federation_utils.full_url(
            "/federation/music/import/batch/{}".format(self.uuid)
        )


class ImportJob(models.Model):
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    replace_if_duplicate = models.BooleanField(default=False)
    batch = models.ForeignKey(
        ImportBatch, related_name="jobs", on_delete=models.CASCADE
    )
    upload = models.ForeignKey(
        Upload, related_name="jobs", null=True, blank=True, on_delete=models.CASCADE
    )
    source = models.CharField(max_length=500)
    mbid = models.UUIDField(editable=False, null=True, blank=True)

    status = models.CharField(
        choices=IMPORT_STATUS_CHOICES, default="pending", max_length=30
    )
    audio_file = models.FileField(
        upload_to="imports/%Y/%m/%d", max_length=255, null=True, blank=True
    )

    library_track = models.ForeignKey(
        "federation.LibraryTrack",
        related_name="import_jobs",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    audio_file_size = models.IntegerField(null=True, blank=True)

    class Meta:
        ordering = ("id",)

    def save(self, **kwargs):
        if self.audio_file and not self.audio_file_size:
            self.audio_file_size = self.audio_file.size
        return super().save(**kwargs)


LIBRARY_PRIVACY_LEVEL_CHOICES = [
    (k, l) for k, l in fields.PRIVACY_LEVEL_CHOICES if k != "followers"
]


class LibraryQuerySet(models.QuerySet):
    def with_follows(self, actor):
        return self.prefetch_related(
            models.Prefetch(
                "received_follows",
                queryset=federation_models.LibraryFollow.objects.filter(actor=actor),
                to_attr="_follows",
            )
        )

    def viewable_by(self, actor):
        from funkwhale_api.federation.models import LibraryFollow

        if actor is None:
            return Library.objects.filter(privacy_level="everyone")

        me_query = models.Q(privacy_level="me", actor=actor)
        instance_query = models.Q(privacy_level="instance", actor__domain=actor.domain)
        followed_libraries = LibraryFollow.objects.filter(
            actor=actor, approved=True
        ).values_list("target", flat=True)
        return Library.objects.filter(
            me_query
            | instance_query
            | models.Q(privacy_level="everyone")
            | models.Q(pk__in=followed_libraries)
        )


class Library(federation_models.FederationMixin):
    uuid = models.UUIDField(unique=True, db_index=True, default=uuid.uuid4)
    actor = models.ForeignKey(
        "federation.Actor", related_name="libraries", on_delete=models.CASCADE
    )
    followers_url = models.URLField(max_length=500)
    creation_date = models.DateTimeField(default=timezone.now)
    name = models.CharField(max_length=100)
    description = models.TextField(max_length=5000, null=True, blank=True)
    privacy_level = models.CharField(
        choices=LIBRARY_PRIVACY_LEVEL_CHOICES, default="me", max_length=25
    )
    uploads_count = models.PositiveIntegerField(default=0)
    objects = LibraryQuerySet.as_manager()

    def get_federation_id(self):
        return federation_utils.full_url(
            reverse("federation:music:libraries-detail", kwargs={"uuid": self.uuid})
        )

    def save(self, **kwargs):
        if not self.pk and not self.fid and self.actor.get_user():
            self.fid = self.get_federation_id()
            self.followers_url = self.fid + "/followers"

        return super().save(**kwargs)

    def should_autoapprove_follow(self, actor):
        if self.privacy_level == "everyone":
            return True
        if self.privacy_level == "instance" and actor.get_user():
            return True
        return False

    def schedule_scan(self, actor, force=False):
        latest_scan = (
            self.scans.exclude(status="errored").order_by("-creation_date").first()
        )
        delay_between_scans = datetime.timedelta(seconds=3600 * 24)
        now = timezone.now()
        if (
            not force
            and latest_scan
            and latest_scan.creation_date + delay_between_scans > now
        ):
            return

        scan = self.scans.create(total_files=self.uploads_count, actor=actor)
        from . import tasks

        common_utils.on_commit(tasks.start_library_scan.delay, library_scan_id=scan.pk)
        return scan


SCAN_STATUS = [
    ("pending", "pending"),
    ("scanning", "scanning"),
    ("errored", "errored"),
    ("finished", "finished"),
]


class LibraryScan(models.Model):
    actor = models.ForeignKey(
        "federation.Actor", null=True, blank=True, on_delete=models.CASCADE
    )
    library = models.ForeignKey(Library, related_name="scans", on_delete=models.CASCADE)
    total_files = models.PositiveIntegerField(default=0)
    processed_files = models.PositiveIntegerField(default=0)
    errored_files = models.PositiveIntegerField(default=0)
    status = models.CharField(default="pending", max_length=25)
    creation_date = models.DateTimeField(default=timezone.now)
    modification_date = models.DateTimeField(null=True, blank=True)


@receiver(post_save, sender=ImportJob)
def update_batch_status(sender, instance, **kwargs):
    instance.batch.update_status()


@receiver(post_save, sender=ImportBatch)
def update_request_status(sender, instance, created, **kwargs):
    update_fields = kwargs.get("update_fields", []) or []
    if not instance.import_request:
        return

    if not created and "status" not in update_fields:
        return

    r_status = instance.import_request.status
    status = instance.status

    if status == "pending" and r_status == "pending":
        # let's mark the request as accepted since we started an import
        instance.import_request.status = "accepted"
        return instance.import_request.save(update_fields=["status"])

    if status == "finished" and r_status == "accepted":
        # let's mark the request as imported since the import is over
        instance.import_request.status = "imported"
        return instance.import_request.save(update_fields=["status"])


@receiver(models.signals.post_save, sender=Album)
def warm_album_covers(sender, instance, **kwargs):
    if not instance.cover or not settings.CREATE_IMAGE_THUMBNAILS:
        return
    album_covers_warmer = VersatileImageFieldWarmer(
        instance_or_queryset=instance, rendition_key_set="square", image_attr="cover"
    )
    num_created, failed_to_create = album_covers_warmer.warm()

           

api/funkwhale_api/playlists/models.py ¶

            from django.db import models
from django.db import transaction
from django.utils import timezone
from funkwhale_api.common import fields
from funkwhale_api.common import preferences
from funkwhale_api.music import models as music_models
from rest_framework import exceptions


class PlaylistQuerySet(models.QuerySet):
    def with_tracks_count(self):
        return self.annotate(_tracks_count=models.Count("playlist_tracks"))

    def with_duration(self):
        return self.annotate(
            duration=models.Sum("playlist_tracks__track__uploads__duration")
        )

    def with_covers(self):
        album_prefetch = models.Prefetch(
            "album", queryset=music_models.Album.objects.only("cover", "artist_id")
        )
        track_prefetch = models.Prefetch(
            "track",
            queryset=music_models.Track.objects.prefetch_related(album_prefetch).only(
                "id", "album_id"
            ),
        )

        plt_prefetch = models.Prefetch(
            "playlist_tracks",
            queryset=PlaylistTrack.objects.all()
            .exclude(track__album__cover=None)
            .exclude(track__album__cover="")
            .order_by("index")
            .only("id", "playlist_id", "track_id")
            .prefetch_related(track_prefetch),
            to_attr="plts_for_cover",
        )
        return self.prefetch_related(plt_prefetch)

    def with_playable_plts(self, actor):
        return self.prefetch_related(
            models.Prefetch(
                "playlist_tracks",
                queryset=PlaylistTrack.objects.playable_by(actor),
                to_attr="playable_plts",
            )
        )

    def playable_by(self, actor, include=True):
        plts = PlaylistTrack.objects.playable_by(actor, include)
        if include:
            return self.filter(playlist_tracks__in=plts).distinct()
        else:
            return self.exclude(playlist_tracks__in=plts).distinct()


class Playlist(models.Model):
    name = models.CharField(max_length=50)
    user = models.ForeignKey(
        "users.User", related_name="playlists", on_delete=models.CASCADE
    )
    creation_date = models.DateTimeField(default=timezone.now)
    modification_date = models.DateTimeField(auto_now=True)
    privacy_level = fields.get_privacy_field()

    objects = PlaylistQuerySet.as_manager()

    def __str__(self):
        return self.name

    @transaction.atomic
    def insert(self, plt, index=None):
        """
        Given a PlaylistTrack, insert it at the correct index in the playlist,
        and update other tracks index if necessary.
        """
        old_index = plt.index
        move = old_index is not None
        if index is not None and index == old_index:
            # moving at same position, just skip
            return index

        existing = self.playlist_tracks.select_for_update()
        if move:
            existing = existing.exclude(pk=plt.pk)
        total = existing.filter(index__isnull=False).count()

        if index is None:
            # we simply increment the last track index by 1
            index = total

        if index > total:
            raise exceptions.ValidationError("Index is not continuous")

        if index < 0:
            raise exceptions.ValidationError("Index must be zero or positive")

        if move:
            # we remove the index temporarily, to avoid integrity errors
            plt.index = None
            plt.save(update_fields=["index"])
            if index > old_index:
                # new index is higher than current, we decrement previous tracks
                to_update = existing.filter(index__gt=old_index, index__lte=index)
                to_update.update(index=models.F("index") - 1)
            if index < old_index:
                # new index is lower than current, we increment next tracks
                to_update = existing.filter(index__lt=old_index, index__gte=index)
                to_update.update(index=models.F("index") + 1)
        else:
            to_update = existing.filter(index__gte=index)
            to_update.update(index=models.F("index") + 1)

        plt.index = index
        plt.save(update_fields=["index"])
        self.save(update_fields=["modification_date"])
        return index

    @transaction.atomic
    def remove(self, index):
        existing = self.playlist_tracks.select_for_update()
        self.save(update_fields=["modification_date"])
        to_update = existing.filter(index__gt=index)
        return to_update.update(index=models.F("index") - 1)

    @transaction.atomic
    def insert_many(self, tracks):
        existing = self.playlist_tracks.select_for_update()
        now = timezone.now()
        total = existing.filter(index__isnull=False).count()
        max_tracks = preferences.get("playlists__max_tracks")
        if existing.count() + len(tracks) > max_tracks:
            raise exceptions.ValidationError(
                "Playlist would reach the maximum of {} tracks".format(max_tracks)
            )
        self.save(update_fields=["modification_date"])
        start = total
        plts = [
            PlaylistTrack(
                creation_date=now, playlist=self, track=track, index=start + i
            )
            for i, track in enumerate(tracks)
        ]
        return PlaylistTrack.objects.bulk_create(plts)


class PlaylistTrackQuerySet(models.QuerySet):
    def for_nested_serialization(self, actor=None):
        tracks = music_models.Track.objects.with_playable_uploads(actor)
        tracks = tracks.select_related("artist", "album__artist")
        return self.prefetch_related(
            models.Prefetch("track", queryset=tracks, to_attr="_prefetched_track")
        )

    def annotate_playable_by_actor(self, actor):
        tracks = (
            music_models.Upload.objects.playable_by(actor)
            .filter(track__pk=models.OuterRef("track"))
            .order_by("id")
            .values("id")[:1]
        )
        subquery = models.Subquery(tracks)
        return self.annotate(is_playable_by_actor=subquery)

    def playable_by(self, actor, include=True):
        tracks = music_models.Track.objects.playable_by(actor, include)
        if include:
            return self.filter(track__pk__in=tracks).distinct()
        else:
            return self.exclude(track__pk__in=tracks).distinct()


class PlaylistTrack(models.Model):
    track = models.ForeignKey(
        "music.Track", related_name="playlist_tracks", on_delete=models.CASCADE
    )
    index = models.PositiveIntegerField(null=True, blank=True)
    playlist = models.ForeignKey(
        Playlist, related_name="playlist_tracks", on_delete=models.CASCADE
    )
    creation_date = models.DateTimeField(default=timezone.now)

    objects = PlaylistTrackQuerySet.as_manager()

    class Meta:
        ordering = ("-playlist", "index")
        unique_together = ("playlist", "index")

    def delete(self, *args, **kwargs):
        playlist = self.playlist
        index = self.index
        update_indexes = kwargs.pop("update_indexes", False)
        r = super().delete(*args, **kwargs)
        if index is not None and update_indexes:
            playlist.remove(index)
        return r

           

api/funkwhale_api/radios/models.py ¶

            from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import JSONField
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models
from django.utils import timezone
from funkwhale_api.music.models import Track

from . import filters


class Radio(models.Model):
    CONFIG_VERSION = 0
    user = models.ForeignKey(
        "users.User",
        related_name="radios",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
    )
    name = models.CharField(max_length=100)
    description = models.TextField(blank=True)
    creation_date = models.DateTimeField(default=timezone.now)
    is_public = models.BooleanField(default=False)
    version = models.PositiveIntegerField(default=0)
    config = JSONField(encoder=DjangoJSONEncoder)

    def get_candidates(self):
        return filters.run(self.config)


class RadioSession(models.Model):
    user = models.ForeignKey(
        "users.User",
        related_name="radio_sessions",
        null=True,
        blank=True,
        on_delete=models.CASCADE,
    )
    session_key = models.CharField(max_length=100, null=True, blank=True)
    radio_type = models.CharField(max_length=50)
    custom_radio = models.ForeignKey(
        Radio, related_name="sessions", null=True, blank=True, on_delete=models.CASCADE
    )
    creation_date = models.DateTimeField(default=timezone.now)
    related_object_content_type = models.ForeignKey(
        ContentType, blank=True, null=True, on_delete=models.CASCADE
    )
    related_object_id = models.PositiveIntegerField(blank=True, null=True)
    related_object = GenericForeignKey(
        "related_object_content_type", "related_object_id"
    )

    def save(self, **kwargs):
        self.radio.clean(self)
        super().save(**kwargs)

    @property
    def next_position(self):
        next_position = 1

        last_session_track = self.session_tracks.all().order_by("-position").first()
        if last_session_track:
            next_position = last_session_track.position + 1

        return next_position

    def add(self, track):
        new_session_track = RadioSessionTrack.objects.create(
            track=track, session=self, position=self.next_position
        )

        return new_session_track

    @property
    def radio(self):
        from .registries import registry

        return registry[self.radio_type](session=self)


class RadioSessionTrack(models.Model):
    session = models.ForeignKey(
        RadioSession, related_name="session_tracks", on_delete=models.CASCADE
    )
    position = models.IntegerField(default=1)
    track = models.ForeignKey(
        Track, related_name="radio_session_tracks", on_delete=models.CASCADE
    )

    class Meta:
        ordering = ("session", "position")
        unique_together = ("session", "position")

           

api/funkwhale_api/requests/models.py ¶

            from django.db import models
from django.utils import timezone

NATURE_CHOICES = [("artist", "artist"), ("album", "album"), ("track", "track")]

STATUS_CHOICES = [
    ("pending", "pending"),
    ("accepted", "accepted"),
    ("imported", "imported"),
    ("closed", "closed"),
]


class ImportRequest(models.Model):
    creation_date = models.DateTimeField(default=timezone.now)
    imported_date = models.DateTimeField(null=True, blank=True)
    user = models.ForeignKey(
        "users.User", related_name="import_requests", on_delete=models.CASCADE
    )
    artist_name = models.CharField(max_length=250)
    albums = models.CharField(max_length=3000, null=True, blank=True)
    status = models.CharField(choices=STATUS_CHOICES, max_length=50, default="pending")
    comment = models.TextField(null=True, blank=True, max_length=3000)

           

api/funkwhale_api/users/models.py ¶

            from __future__ import absolute_import
from __future__ import unicode_literals

import binascii
import datetime
import os
import random
import string
import uuid

from django.conf import settings
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.dispatch import receiver
from django.urls import reverse
from django.utils import timezone
from django.utils.encoding import python_2_unicode_compatible
from django.utils.translation import ugettext_lazy as _
from django_auth_ldap.backend import populate_user as ldap_populate_user
from funkwhale_api.common import fields
from funkwhale_api.common import preferences
from funkwhale_api.common import utils as common_utils
from funkwhale_api.common import validators as common_validators
from funkwhale_api.federation import keys
from funkwhale_api.federation import models as federation_models
from funkwhale_api.federation import utils as federation_utils
from versatileimagefield.fields import VersatileImageField
from versatileimagefield.image_warmer import VersatileImageFieldWarmer


def get_token():
    return binascii.b2a_hex(os.urandom(15)).decode("utf-8")


PERMISSIONS_CONFIGURATION = {
    "moderation": {
        "label": "Moderation",
        "help_text": "Block/mute/remove domains, users and content",
    },
    "library": {
        "label": "Manage library",
        "help_text": "Manage library, delete files, tracks, artists, albums...",
    },
    "settings": {"label": "Manage instance-level settings", "help_text": ""},
}

PERMISSIONS = sorted(PERMISSIONS_CONFIGURATION.keys())


get_file_path = common_utils.ChunkedPath("users/avatars", preserve_file_name=False)


@python_2_unicode_compatible
class User(AbstractUser):

    # First Name and Last Name do not cover name patterns
    # around the globe.
    name = models.CharField(_("Name of User"), blank=True, max_length=255)

    # updated on logout or password change, to invalidate JWT
    secret_key = models.UUIDField(default=uuid.uuid4, null=True)
    privacy_level = fields.get_privacy_field()

    # Unfortunately, Subsonic API assumes a MD5/password authentication
    # scheme, which is weak in terms of security, and not achievable
    # anyway since django use stronger schemes for storing passwords.
    # Users that want to use the subsonic API from external client
    # should set this token and use it as their password in such clients
    subsonic_api_token = models.CharField(blank=True, null=True, max_length=255)

    # permissions
    permission_moderation = models.BooleanField(
        PERMISSIONS_CONFIGURATION["moderation"]["label"],
        help_text=PERMISSIONS_CONFIGURATION["moderation"]["help_text"],
        default=False,
    )
    permission_library = models.BooleanField(
        PERMISSIONS_CONFIGURATION["library"]["label"],
        help_text=PERMISSIONS_CONFIGURATION["library"]["help_text"],
        default=False,
    )
    permission_settings = models.BooleanField(
        PERMISSIONS_CONFIGURATION["settings"]["label"],
        help_text=PERMISSIONS_CONFIGURATION["settings"]["help_text"],
        default=False,
    )

    last_activity = models.DateTimeField(default=None, null=True, blank=True)

    invitation = models.ForeignKey(
        "Invitation",
        related_name="users",
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    avatar = VersatileImageField(
        upload_to=get_file_path,
        null=True,
        blank=True,
        max_length=150,
        validators=[
            common_validators.ImageDimensionsValidator(min_width=50, min_height=50),
            common_validators.FileValidator(
                allowed_extensions=["png", "jpg", "jpeg", "gif"],
                max_size=1024 * 1024 * 2,
            ),
        ],
    )
    actor = models.OneToOneField(
        "federation.Actor",
        related_name="user",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )

    upload_quota = models.PositiveIntegerField(null=True, blank=True)

    def __str__(self):
        return self.username

    def get_permissions(self, defaults=None):
        defaults = defaults or preferences.get("users__default_permissions")
        perms = {}
        for p in PERMISSIONS:
            v = (
                self.is_superuser
                or getattr(self, "permission_{}".format(p))
                or p in defaults
            )
            perms[p] = v
        return perms

    @property
    def all_permissions(self):
        return self.get_permissions()

    def has_permissions(self, *perms, **kwargs):
        operator = kwargs.pop("operator", "and")
        if operator not in ["and", "or"]:
            raise ValueError("Invalid operator {}".format(operator))
        permissions = self.get_permissions()
        checker = all if operator == "and" else any
        return checker([permissions[p] for p in perms])

    def get_absolute_url(self):
        return reverse("users:detail", kwargs={"username": self.username})

    def update_secret_key(self):
        self.secret_key = uuid.uuid4()
        return self.secret_key

    def update_subsonic_api_token(self):
        self.subsonic_api_token = get_token()
        return self.subsonic_api_token

    def set_password(self, raw_password):
        super().set_password(raw_password)
        self.update_secret_key()
        if self.subsonic_api_token:
            self.update_subsonic_api_token()

    def get_activity_url(self):
        return settings.FUNKWHALE_URL + "/@{}".format(self.username)

    def record_activity(self):
        """
        Simply update the last_activity field if current value is too old
        than a threshold. This is useful to keep a track of inactive accounts.
        """
        current = self.last_activity
        delay = 60 * 15  # fifteen minutes
        now = timezone.now()

        if current is None or current < now - datetime.timedelta(seconds=delay):
            self.last_activity = now
            self.save(update_fields=["last_activity"])

    def create_actor(self):
        self.actor = create_actor(self)
        self.save(update_fields=["actor"])
        return self.actor

    def get_upload_quota(self):
        return self.upload_quota or preferences.get("users__upload_quota")

    def get_quota_status(self):
        data = self.actor.get_current_usage()
        max_ = self.get_upload_quota()
        return {
            "max": max_,
            "remaining": max(max_ - (data["total"] / 1000 / 1000), 0),
            "current": data["total"] / 1000 / 1000,
            "skipped": data["skipped"] / 1000 / 1000,
            "pending": data["pending"] / 1000 / 1000,
            "finished": data["finished"] / 1000 / 1000,
            "errored": data["errored"] / 1000 / 1000,
        }

    def get_channels_groups(self):
        groups = ["imports", "inbox"]

        return ["user.{}.{}".format(self.pk, g) for g in groups]

    def full_username(self):
        return "{}@{}".format(self.username, settings.FEDERATION_HOSTNAME)


def generate_code(length=10):
    return "".join(
        random.SystemRandom().choice(string.ascii_uppercase) for _ in range(length)
    )


class InvitationQuerySet(models.QuerySet):
    def open(self, include=True):
        now = timezone.now()
        qs = self.annotate(_users=models.Count("users"))
        query = models.Q(_users=0, expiration_date__gt=now)
        if include:
            return qs.filter(query)
        return qs.exclude(query)


class Invitation(models.Model):
    creation_date = models.DateTimeField(default=timezone.now)
    expiration_date = models.DateTimeField()
    owner = models.ForeignKey(
        User, related_name="invitations", on_delete=models.CASCADE
    )
    code = models.CharField(max_length=50, unique=True)

    objects = InvitationQuerySet.as_manager()

    def save(self, **kwargs):
        if not self.code:
            self.code = generate_code()
        if not self.expiration_date:
            self.expiration_date = self.creation_date + datetime.timedelta(
                days=settings.USERS_INVITATION_EXPIRATION_DAYS
            )

        return super().save(**kwargs)


def get_actor_data(username):
    slugified_username = federation_utils.slugify_username(username)
    return {
        "preferred_username": slugified_username,
        "domain": federation_models.Domain.objects.get_or_create(
            name=settings.FEDERATION_HOSTNAME
        )[0],
        "type": "Person",
        "name": username,
        "manually_approves_followers": False,
        "fid": federation_utils.full_url(
            reverse(
                "federation:actors-detail",
                kwargs={"preferred_username": slugified_username},
            )
        ),
        "shared_inbox_url": federation_models.get_shared_inbox_url(),
        "inbox_url": federation_utils.full_url(
            reverse(
                "federation:actors-inbox",
                kwargs={"preferred_username": slugified_username},
            )
        ),
        "outbox_url": federation_utils.full_url(
            reverse(
                "federation:actors-outbox",
                kwargs={"preferred_username": slugified_username},
            )
        ),
        "followers_url": federation_utils.full_url(
            reverse(
                "federation:actors-followers",
                kwargs={"preferred_username": slugified_username},
            )
        ),
        "following_url": federation_utils.full_url(
            reverse(
                "federation:actors-following",
                kwargs={"preferred_username": slugified_username},
            )
        ),
    }


def create_actor(user):
    args = get_actor_data(user.username)
    private, public = keys.get_key_pair()
    args["private_key"] = private.decode("utf-8")
    args["public_key"] = public.decode("utf-8")

    return federation_models.Actor.objects.create(user=user, **args)


@receiver(ldap_populate_user)
def init_ldap_user(sender, user, ldap_user, **kwargs):
    if not user.actor:
        user.actor = create_actor(user)


@receiver(models.signals.post_save, sender=User)
def warm_user_avatar(sender, instance, **kwargs):
    if not instance.avatar or not settings.CREATE_IMAGE_THUMBNAILS:
        return
    user_avatar_warmer = VersatileImageFieldWarmer(
        instance_or_queryset=instance, rendition_key_set="square", image_attr="avatar"
    )
    num_created, failed_to_create = user_avatar_warmer.warm()

           