Compare commits

...

25 Commits

Author SHA1 Message Date
908613306f
ci: add linter action 2024-05-22 01:31:58 +08:00
38e0d7f8d1
ci: use ruff as formatter and linter 2024-05-22 01:23:55 +08:00
937bbe2eee
ci: run tests on every branch 2024-05-22 00:29:41 +08:00
e7398be07e
chore: update README
notice about v0.3.0 development
2024-04-04 18:42:58 +08:00
64285c350c
ci: dist file discovery 2024-04-01 00:10:30 +08:00
62c3431cff
chore: v0.2.2 2024-03-31 23:56:18 +08:00
280543660a
ci: build package using github actions 2024-03-31 23:54:58 +08:00
8dc433b12a
feat: ETERNAL rating class support 2024-03-20 15:52:55 +08:00
2bd64bbd5e
feat(external): supporting chart info database 2024-03-16 01:54:26 +08:00
54749c8df2
feat(db): DEF v2 scores export supporting 2024-02-27 17:23:43 +08:00
36364d6e3d
ci: pytest actions 2023-11-09 17:31:55 +08:00
b14c3e82b4
test: add tests 2023-11-09 16:33:51 +08:00
14f4cef426
chore: fix pylint warnings 2023-11-02 01:10:31 +08:00
92fcc53015
fix: typing issues 2023-11-02 00:21:33 +08:00
0764308638
chore: v0.2.1 2023-11-01 14:09:44 +08:00
e5c1e0ef4a
fix(external): fix_timestamp digits counting (#3) 2023-10-27 00:23:17 +08:00
8c48d76c65
feat(external): add support for smartrte.github.io 2023-10-21 21:38:47 +08:00
d79c73df8c
impr(external): Andreal B30 overflow data 2023-10-21 20:58:40 +08:00
7a64ec4a4a
fix(models): null value checking 2023-10-21 19:03:21 +08:00
a9f8ba6e22
impr(external): St3ScoreParser improvements 2023-10-18 01:21:27 +08:00
13ea4d9e97
fix(external): fix_timestamp current timestamp digits counting 2023-10-18 01:17:12 +08:00
24d46e4615
feat(external): Online parser 2023-10-18 01:08:48 +08:00
62c85e9e82
fix(models): null value checking 2023-10-16 01:14:34 +08:00
c7de60ee03
feat(utils): Kanae day/night util 2023-10-12 17:05:33 +08:00
7c000d01cb
feat(utils): text converters 2023-10-12 16:21:37 +08:00
40 changed files with 701 additions and 91 deletions

View File

@ -0,0 +1,48 @@
name: "Build and draft a release"
on:
workflow_dispatch:
push:
tags:
- "v[0-9]+.[0-9]+.[0-9]+"
permissions:
contents: write
discussions: write
jobs:
build-and-draft-release:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up Python environment
uses: actions/setup-python@v5
with:
python-version: "3.x"
- name: Build package
run: |
pip install build
python -m build
- name: Remove `v` in tag name
uses: mad9000/actions-find-and-replace-string@5
id: tagNameReplaced
with:
source: ${{ github.ref_name }}
find: "v"
replace: ""
- name: Draft a release
uses: softprops/action-gh-release@v2
with:
discussion_category_name: New releases
draft: true
generate_release_notes: true
files: |
dist/arcaea_offline-${{ steps.tagNameReplaced.outputs.value }}*.whl
dist/arcaea-offline-${{ steps.tagNameReplaced.outputs.value }}.tar.gz

40
.github/workflows/main.yml vendored Normal file
View File

@ -0,0 +1,40 @@
name: test & lint
on:
push:
branches:
- '*'
pull_request:
types: [opened, reopened]
workflow_dispatch:
jobs:
pytest:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dev dependencies
run: 'pip install .[dev]'
- name: Run tests
run: 'pytest -v'
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dev dependencies
run: 'pip install .[dev]'
- name: Run linter
run: 'ruff check'

View File

@ -4,11 +4,10 @@ repos:
hooks: hooks:
- id: end-of-file-fixer - id: end-of-file-fixer
- id: trailing-whitespace - id: trailing-whitespace
- repo: https://github.com/psf/black
rev: 23.1.0 - repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.4.4
hooks: hooks:
- id: black - id: ruff
- repo: https://github.com/PyCQA/isort args: ["--fix"]
rev: 5.12.0 - id: ruff-format
hooks:
- id: isort

2
.sourcery.yaml Normal file
View File

@ -0,0 +1,2 @@
rule_settings:
python_version: '3.8'

View File

@ -10,9 +10,14 @@ English | [简体中文](./README.zh_Hans.md)
## WIP ## WIP
> **Warning** > [!CAUTION]
> This project is under active development, thus it is unstable and API may change frequently. > This project is under active development, thus it is unstable and API may change frequently.
> [!IMPORTANT]
> v0.3.0 is under development, check out [this branch](https://github.com/283375/arcaea-offline/tree/0.3.0-refactor)!
>
> Once v0.3.0 is ready for release, this repository will be transferred to *[ArcaeaOffline](https://github.com/ArcaeaOffline)/core-python*
## What is this ## What is this
This is the core library of `Arcaea Offline`, designed to manage player scores, calculate their potential, and provide various useful tools. This is the core library of `Arcaea Offline`, designed to manage player scores, calculate their potential, and provide various useful tools.

View File

@ -8,9 +8,14 @@
## WIP ## WIP
> **Warning** > [!CAUTION]
> 该项目正处于早期开发阶段,不能保证稳定性,且 API 可能随时变动。 > 该项目正处于早期开发阶段,不能保证稳定性,且 API 可能随时变动。
> [!IMPORTANT]
> v0.3.0 正在[此分支](https://github.com/283375/arcaea-offline/tree/0.3.0-refactor)下开发!
>
> 在 v0.3.0 准备好发布后,此存储库将被迁移至 *[ArcaeaOffline](https://github.com/ArcaeaOffline)/core-python*
## 这是什么? ## 这是什么?
这是 `Arcaea Offline` 的核心依赖库,用于维护分数数据库、计算潜力值,并提供一些实用工具。 这是 `Arcaea Offline` 的核心依赖库,用于维护分数数据库、计算潜力值,并提供一些实用工具。

View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "arcaea-offline" name = "arcaea-offline"
version = "0.2.0" version = "0.2.2"
authors = [{ name = "283375", email = "log_283375@163.com" }] authors = [{ name = "283375", email = "log_283375@163.com" }]
description = "Manage your local Arcaea score database." description = "Manage your local Arcaea score database."
readme = "README.md" readme = "README.md"
@ -20,13 +20,34 @@ classifiers = [
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
] ]
[project.optional-dependencies]
dev = ["ruff~=0.4", "pre-commit~=3.3", "pytest~=7.4", "tox~=4.11"]
[project.urls] [project.urls]
"Homepage" = "https://github.com/283375/arcaea-offline" "Homepage" = "https://github.com/283375/arcaea-offline"
"Bug Tracker" = "https://github.com/283375/arcaea-offline/issues" "Bug Tracker" = "https://github.com/283375/arcaea-offline/issues"
[tool.isort]
profile = "black"
src_paths = ["src/arcaea_offline"]
[tool.pyright] [tool.pyright]
ignore = ["**/__debug*.*"] ignore = ["build/"]
[tool.ruff.lint]
# Full list: https://docs.astral.sh/ruff/rules
select = [
"E", # pycodestyle (Error)
"W", # pycodestyle (Warning)
"F", # pyflakes
"I", # isort
"PL", # pylint
"N", # pep8-naming
"FBT", # flake8-boolean-trap
"A", # flake8-builtins
"DTZ", # flake8-datetimez
"LOG", # flake8-logging
"Q", # flake8-quotes
"G", # flake8-logging-format
"PIE", # flake8-pie
"PT", # flake8-pytest-style
]
ignore = [
"E501", # line-too-long
]

View File

@ -1,3 +1,4 @@
black==23.3.0 ruff~=0.4
isort==5.12.0 pre-commit~=3.3
pre-commit==3.3.1 pytest~=7.4
tox~=4.11

View File

@ -16,10 +16,9 @@ def calculate_score_range(notes: int, pure: int, far: int):
def calculate_score_modifier(score: int) -> Decimal: def calculate_score_modifier(score: int) -> Decimal:
if score >= 10000000: if score >= 10000000:
return Decimal(2) return Decimal(2)
elif score >= 9800000: if score >= 9800000:
return Decimal(1) + (Decimal(score - 9800000) / 200000) return Decimal(1) + (Decimal(score - 9800000) / 200000)
else: return Decimal(score - 9500000) / 300000
return Decimal(score - 9500000) / 300000
def calculate_play_rating(constant: int, score: int) -> Decimal: def calculate_play_rating(constant: int, score: int) -> Decimal:
@ -35,6 +34,7 @@ def calculate_shiny_pure(notes: int, score: int, pure: int, far: int) -> int:
@dataclass @dataclass
class ConstantsFromPlayRatingResult: class ConstantsFromPlayRatingResult:
# pylint: disable=invalid-name
EXPlus: Tuple[Decimal, Decimal] EXPlus: Tuple[Decimal, Decimal]
EX: Tuple[Decimal, Decimal] EX: Tuple[Decimal, Decimal]
AA: Tuple[Decimal, Decimal] AA: Tuple[Decimal, Decimal]
@ -44,10 +44,12 @@ class ConstantsFromPlayRatingResult:
def calculate_constants_from_play_rating(play_rating: Union[Decimal, str, float, int]): def calculate_constants_from_play_rating(play_rating: Union[Decimal, str, float, int]):
# pylint: disable=no-value-for-parameter
play_rating = Decimal(play_rating) play_rating = Decimal(play_rating)
ranges = [] ranges = []
for upperScore, lowerScore in [ for upper_score, lower_score in [
(10000000, 9900000), (10000000, 9900000),
(9899999, 9800000), (9899999, 9800000),
(9799999, 9500000), (9799999, 9500000),
@ -55,10 +57,10 @@ def calculate_constants_from_play_rating(play_rating: Union[Decimal, str, float,
(9199999, 8900000), (9199999, 8900000),
(8899999, 8600000), (8899999, 8600000),
]: ]:
upperScoreModifier = calculate_score_modifier(upperScore) upper_score_modifier = calculate_score_modifier(upper_score)
lowerScoreModifier = calculate_score_modifier(lowerScore) lower_score_modifier = calculate_score_modifier(lower_score)
ranges.append( ranges.append(
(play_rating - upperScoreModifier, play_rating - lowerScoreModifier) (play_rating - upper_score_modifier, play_rating - lower_score_modifier)
) )
return ConstantsFromPlayRatingResult(*ranges) return ConstantsFromPlayRatingResult(*ranges)

View File

@ -97,9 +97,8 @@ class LegacyMapStepBooster(StepBooster):
def final_value(self) -> Decimal: def final_value(self) -> Decimal:
stamina_multiplier = Decimal(self.stamina) stamina_multiplier = Decimal(self.stamina)
if self.fragments is None: fragments_multiplier = Decimal(1)
fragments_multiplier = Decimal(1) if self.fragments == 100:
elif self.fragments == 100:
fragments_multiplier = Decimal("1.1") fragments_multiplier = Decimal("1.1")
elif self.fragments == 250: elif self.fragments == 250:
fragments_multiplier = Decimal("1.25") fragments_multiplier = Decimal("1.25")
@ -118,7 +117,7 @@ def calculate_step_original(
*, *,
partner_bonus: Optional[PartnerBonus] = None, partner_bonus: Optional[PartnerBonus] = None,
step_booster: Optional[StepBooster] = None, step_booster: Optional[StepBooster] = None,
): ) -> Decimal:
ptt = play_result.play_rating ptt = play_result.play_rating
step = play_result.partner_step step = play_result.partner_step
if partner_bonus: if partner_bonus:
@ -128,13 +127,13 @@ def calculate_step_original(
partner_bonus_step = Decimal("0") partner_bonus_step = Decimal("0")
partner_bonus_multiplier = Decimal("1.0") partner_bonus_multiplier = Decimal("1.0")
play_result = (Decimal("2.45") * ptt.sqrt() + Decimal("2.5")) * (step / 50) result = (Decimal("2.45") * ptt.sqrt() + Decimal("2.5")) * (step / 50)
play_result += partner_bonus_step result += partner_bonus_step
play_result *= partner_bonus_multiplier result *= partner_bonus_multiplier
if step_booster: if step_booster:
play_result *= step_booster.final_value() result *= step_booster.final_value()
return play_result return result
def calculate_step( def calculate_step(
@ -142,12 +141,12 @@ def calculate_step(
*, *,
partner_bonus: Optional[PartnerBonus] = None, partner_bonus: Optional[PartnerBonus] = None,
step_booster: Optional[StepBooster] = None, step_booster: Optional[StepBooster] = None,
): ) -> Decimal:
play_result_original = calculate_step_original( result_original = calculate_step_original(
play_result, partner_bonus=partner_bonus, step_booster=step_booster play_result, partner_bonus=partner_bonus, step_booster=step_booster
) )
return round(play_result_original, 1) return round(result_original, 1)
def calculate_play_rating_from_step( def calculate_play_rating_from_step(

View File

@ -5,12 +5,29 @@ from typing import Iterable, List, Optional, Type, Union
from sqlalchemy import Engine, func, inspect, select from sqlalchemy import Engine, func, inspect, select
from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, sessionmaker from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, sessionmaker
from .calculate import calculate_score_modifier
from .external.arcsong.arcsong_json import ArcSongJsonBuilder from .external.arcsong.arcsong_json import ArcSongJsonBuilder
from .external.exports import ScoreExport, exporters from .external.exports import ArcaeaOfflineDEFV2_Score, ScoreExport, exporters
from .models.config import * from .models.config import ConfigBase, Property
from .models.scores import * from .models.scores import (
from .models.songs import * CalculatedPotential,
Score,
ScoreBest,
ScoreCalculated,
ScoresBase,
ScoresViewBase,
)
from .models.songs import (
Chart,
ChartInfo,
Difficulty,
DifficultyLocalized,
Pack,
PackLocalized,
Song,
SongLocalized,
SongsBase,
SongsViewBase,
)
from .singleton import Singleton from .singleton import Singleton
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -27,18 +44,21 @@ class Database(metaclass=Singleton):
if isinstance(self.engine, Engine): if isinstance(self.engine, Engine):
return return
raise ValueError("No sqlalchemy.Engine instance specified before.") raise ValueError("No sqlalchemy.Engine instance specified before.")
elif isinstance(engine, Engine):
if isinstance(self.engine, Engine): if not isinstance(engine, Engine):
logger.warning(
f"A sqlalchemy.Engine instance {self.engine} has been specified "
f"and will be replaced to {engine}"
)
self.engine = engine
else:
raise ValueError( raise ValueError(
f"A sqlalchemy.Engine instance expected, not {repr(engine)}" f"A sqlalchemy.Engine instance expected, not {repr(engine)}"
) )
if isinstance(self.engine, Engine):
logger.warning(
"A sqlalchemy.Engine instance %r has been specified "
"and will be replaced to %r",
self.engine,
engine,
)
self.engine = engine
@property @property
def engine(self) -> Engine: def engine(self) -> Engine:
return self.__engine # type: ignore return self.__engine # type: ignore
@ -60,7 +80,8 @@ class Database(metaclass=Singleton):
# create tables & views # create tables & views
if checkfirst: if checkfirst:
# > https://github.com/kvesteri/sqlalchemy-utils/issues/396 # > https://github.com/kvesteri/sqlalchemy-utils/issues/396
# > view.create_view() causes DuplicateTableError on Base.metadata.create_all(checkfirst=True) # > view.create_view() causes DuplicateTableError on
# > Base.metadata.create_all(checkfirst=True)
# so if `checkfirst` is True, drop these views before creating # so if `checkfirst` is True, drop these views before creating
SongsViewBase.metadata.drop_all(self.engine) SongsViewBase.metadata.drop_all(self.engine)
ScoresViewBase.metadata.drop_all(self.engine) ScoresViewBase.metadata.drop_all(self.engine)
@ -389,6 +410,15 @@ class Database(metaclass=Singleton):
scores = self.get_scores() scores = self.get_scores()
return [exporters.score(score) for score in scores] return [exporters.score(score) for score in scores]
def export_scores_def_v2(self) -> ArcaeaOfflineDEFV2_Score:
scores = self.get_scores()
return {
"$schema": "https://arcaeaoffline.sevive.xyz/schemas/def/v2/score.schema.json",
"type": "score",
"version": 2,
"scores": [exporters.score_def_v2(score) for score in scores],
}
def generate_arcsong(self): def generate_arcsong(self):
with self.sessionmaker() as session: with self.sessionmaker() as session:
arcsong = ArcSongJsonBuilder(session).generate_arcsong_json() arcsong = ArcSongJsonBuilder(session).generate_arcsong_json()

View File

@ -84,11 +84,15 @@ class AndrealImageGeneratorApiDataConverter:
raise ValueError("No score available.") raise ValueError("No score available.")
best30_avg = self.session.scalar(select(CalculatedPotential.b30)) best30_avg = self.session.scalar(select(CalculatedPotential.b30))
best30_overflow = (
[self.score(score) for score in scores[30:40]] if len(scores) > 30 else []
)
return { return {
"content": { "content": {
"account_info": self.account_info(), "account_info": self.account_info(),
"best30_avg": best30_avg, "best30_avg": best30_avg,
"best30_list": [self.score(score) for score in scores[:30]], "best30_list": [self.score(score) for score in scores[:30]],
"best30_overflow": [self.score(score) for score in scores[-10:]], "best30_overflow": best30_overflow,
} }
} }

View File

@ -1,3 +1,4 @@
from .online import ArcaeaOnlineParser
from .packlist import PacklistParser from .packlist import PacklistParser
from .songlist import SonglistDifficultiesParser, SonglistParser from .songlist import SonglistDifficultiesParser, SonglistParser
from .st3 import St3ScoreParser from .st3 import St3ScoreParser

View File

@ -1,18 +1,45 @@
import contextlib import contextlib
import json import json
import math
import time
from os import PathLike from os import PathLike
from typing import Any, List, Optional, Union from typing import Any, List, Optional, Union
from sqlalchemy.orm import DeclarativeBase, Session from sqlalchemy.orm import DeclarativeBase, Session
def fix_timestamp(timestamp: int) -> Union[int, None]:
"""
Some of the `date` column in st3 are strangely truncated. For example,
a `1670283375` may be truncated to `167028`, even `1`. Yes, a single `1`.
To properly handle this situation, we check the timestamp's digits.
If `digits < 5`, we treat this timestamp as a `None`. Otherwise, we try to
fix the timestamp.
:param timestamp: a POSIX timestamp
:return: `None` if the timestamp's digits < 5, otherwise a fixed POSIX timestamp
"""
# find digit length from https://stackoverflow.com/a/2189827/16484891
# CC BY-SA 2.5
# this might give incorrect result when timestamp > 999999999999997,
# see https://stackoverflow.com/a/28883802/16484891 (CC BY-SA 4.0).
# but that's way too later than 9999-12-31 23:59:59, 253402271999,
# I don't think Arcaea would still be an active updated game by then.
# so don't mind those small issues, just use this.
digits = int(math.log10(abs(timestamp))) + 1 if timestamp != 0 else 1
if digits < 5:
return None
timestamp_str = str(timestamp)
current_timestamp_digits = int(math.log10(int(time.time()))) + 1
timestamp_str = timestamp_str.ljust(current_timestamp_digits, "0")
return int(timestamp_str, 10)
def to_db_value(val: Any) -> Any: def to_db_value(val: Any) -> Any:
if not val: if not val:
return None return None
elif isinstance(val, list): return json.dumps(val, ensure_ascii=False) if isinstance(val, list) else val
return json.dumps(val, ensure_ascii=False)
else:
return val
def is_localized(item: dict, key: str, append_localized: bool = True): def is_localized(item: dict, key: str, append_localized: bool = True):
@ -56,7 +83,7 @@ class ArcaeaParser:
# or maybe a `pathlib.Path` subset # or maybe a `pathlib.Path` subset
# or an `importlib.resources.abc.Traversable` like object # or an `importlib.resources.abc.Traversable` like object
# e.g. `zipfile.Path` # e.g. `zipfile.Path`
file_handle = self.filepath.open(mode="r", encoding="utf-8") file_handle = self.filepath.open(mode="r", encoding="utf-8") # type: ignore
except Exception as e: except Exception as e:
raise ValueError("Invalid `filepath`.") from e raise ValueError("Invalid `filepath`.") from e
@ -64,7 +91,7 @@ class ArcaeaParser:
return file_handle.read() return file_handle.read()
def parse(self) -> List[DeclarativeBase]: def parse(self) -> List[DeclarativeBase]:
... raise NotImplementedError()
def write_database(self, session: Session): def write_database(self, session: Session):
results = self.parse() results = self.parse()

View File

@ -0,0 +1,72 @@
import json
import logging
from datetime import datetime
from typing import Dict, List, Literal, Optional, TypedDict
from ...models import Score
from .common import ArcaeaParser, fix_timestamp
logger = logging.getLogger(__name__)
class TWebApiRatingMeScoreItem(TypedDict):
song_id: str
difficulty: int
modifier: int
rating: float
score: int
perfect_count: int
near_count: int
miss_count: int
clear_type: int
title: Dict[Literal["ja", "en"], str]
artist: str
time_played: int
bg: str
class TWebApiRatingMeValue(TypedDict):
best_rated_scores: List[TWebApiRatingMeScoreItem]
recent_rated_scores: List[TWebApiRatingMeScoreItem]
class TWebApiRatingMeResult(TypedDict):
success: bool
error_code: Optional[int]
value: Optional[TWebApiRatingMeValue]
class ArcaeaOnlineParser(ArcaeaParser):
def parse(self) -> List[Score]:
api_result_root: TWebApiRatingMeResult = json.loads(self.read_file_text())
api_result_value = api_result_root.get("value")
if not api_result_value:
error_code = api_result_root.get("error_code")
raise ValueError(f"Cannot parse API result, error code {error_code}")
best30_score_items = api_result_value.get("best_rated_scores", [])
recent_score_items = api_result_value.get("recent_rated_scores", [])
score_items = best30_score_items + recent_score_items
date_text = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
results: List[Score] = []
for score_item in score_items:
score = Score()
score.song_id = score_item["song_id"]
score.rating_class = score_item["difficulty"]
score.score = score_item["score"]
score.pure = score_item["perfect_count"]
score.far = score_item["near_count"]
score.lost = score_item["miss_count"]
score.date = fix_timestamp(int(score_item["time_played"] / 1000))
score.modifier = score_item["modifier"]
score.clear_type = score_item["clear_type"]
if score.lost == 0:
score.max_recall = score.pure + score.far
score.comment = f"Parsed from web API at {date_text}"
results.append(score)
return results

View File

@ -6,9 +6,6 @@ from .common import ArcaeaParser, is_localized, set_model_localized_attrs
class PacklistParser(ArcaeaParser): class PacklistParser(ArcaeaParser):
def __init__(self, filepath):
super().__init__(filepath)
def parse(self) -> List[Union[Pack, PackLocalized]]: def parse(self) -> List[Union[Pack, PackLocalized]]:
packlist_json_root = json.loads(self.read_file_text()) packlist_json_root = json.loads(self.read_file_text())

View File

@ -6,9 +6,6 @@ from .common import ArcaeaParser, is_localized, set_model_localized_attrs, to_db
class SonglistParser(ArcaeaParser): class SonglistParser(ArcaeaParser):
def __init__(self, filepath):
super().__init__(filepath)
def parse( def parse(
self, self,
) -> List[Union[Song, SongLocalized, Difficulty, DifficultyLocalized]]: ) -> List[Union[Song, SongLocalized, Difficulty, DifficultyLocalized]]:
@ -61,9 +58,6 @@ class SonglistParser(ArcaeaParser):
class SonglistDifficultiesParser(ArcaeaParser): class SonglistDifficultiesParser(ArcaeaParser):
def __init__(self, filepath):
self.filepath = filepath
def parse(self) -> List[Union[Difficulty, DifficultyLocalized]]: def parse(self) -> List[Union[Difficulty, DifficultyLocalized]]:
songlist_json_root = json.loads(self.read_file_text()) songlist_json_root = json.loads(self.read_file_text())

View File

@ -6,21 +6,19 @@ from sqlalchemy import select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from ...models.scores import Score from ...models.scores import Score
from .common import ArcaeaParser from .common import ArcaeaParser, fix_timestamp
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class St3ScoreParser(ArcaeaParser): class St3ScoreParser(ArcaeaParser):
def __init__(self, filepath):
super().__init__(filepath)
def parse(self) -> List[Score]: def parse(self) -> List[Score]:
items = [] items = []
with sqlite3.connect(self.filepath) as st3_conn: with sqlite3.connect(self.filepath) as st3_conn:
cursor = st3_conn.cursor() cursor = st3_conn.cursor()
db_scores = cursor.execute( db_scores = cursor.execute(
"SELECT songId, songDifficulty, score, perfectCount, nearCount, missCount, date, modifier FROM scores" "SELECT songId, songDifficulty, score, perfectCount, nearCount, missCount, "
"date, modifier FROM scores"
).fetchall() ).fetchall()
for ( for (
song_id, song_id,
@ -37,9 +35,6 @@ class St3ScoreParser(ArcaeaParser):
(song_id, rating_class), (song_id, rating_class),
).fetchone()[0] ).fetchone()[0]
date_str = str(date)
date = None if len(date_str) < 7 else int(date_str.ljust(10, "0"))
items.append( items.append(
Score( Score(
song_id=song_id, song_id=song_id,
@ -48,10 +43,10 @@ class St3ScoreParser(ArcaeaParser):
pure=pure, pure=pure,
far=far, far=far,
lost=lost, lost=lost,
date=date, date=fix_timestamp(date),
modifier=modifier, modifier=modifier,
clear_type=clear_type, clear_type=clear_type,
comment="Imported from st3", comment="Parsed from st3",
) )
) )
@ -70,8 +65,9 @@ class St3ScoreParser(ArcaeaParser):
if query_score and skip_duplicate: if query_score and skip_duplicate:
logger.info( logger.info(
f"{repr(parsed_score)} skipped because " "%r skipped because potential duplicate item %r found.",
f"potential duplicate item {repr(query_score)} found." parsed_score,
query_score,
) )
continue continue
session.add(parsed_score) session.add(parsed_score)

View File

@ -122,7 +122,9 @@ class ArcSongJsonBuilder:
pack = self.session.scalar(select(Pack).where(Pack.id == song.set)) pack = self.session.scalar(select(Pack).where(Pack.id == song.set))
if not pack: if not pack:
logger.warning(f'Cannot find pack "{song.set}", using placeholder instead.') logger.warning(
'Cannot find pack "%s", using placeholder instead.', song.set
)
pack = Pack(id="unknown", name="Unknown", description="__PLACEHOLDER__") pack = Pack(id="unknown", name="Unknown", description="__PLACEHOLDER__")
song_localized = self.session.scalar( song_localized = self.session.scalar(
select(SongLocalized).where(SongLocalized.id == song.id) select(SongLocalized).where(SongLocalized.id == song.id)

View File

@ -0,0 +1 @@
from .parser import ChartInfoDbParser

View File

@ -0,0 +1,35 @@
import contextlib
import sqlite3
from typing import List
from sqlalchemy.orm import Session
from ...models.songs import ChartInfo
class ChartInfoDbParser:
def __init__(self, filepath):
self.filepath = filepath
def parse(self) -> List[ChartInfo]:
results = []
with sqlite3.connect(self.filepath) as conn:
with contextlib.closing(conn.cursor()) as cursor:
db_results = cursor.execute(
"SELECT song_id, rating_class, constant, notes FROM charts_info"
).fetchall()
for result in db_results:
chart = ChartInfo(
song_id=result[0],
rating_class=result[1],
constant=result[2],
notes=result[3] or None,
)
results.append(chart)
return results
def write_database(self, session: Session):
results = self.parse()
for result in results:
session.merge(result)

View File

@ -1,2 +1,2 @@
from . import exporters from . import exporters
from .types import ScoreExport from .types import ArcaeaOfflineDEFV2_Score, ScoreExport

View File

@ -1,5 +1,5 @@
from ...models import Score from ...models import Score
from .types import ScoreExport from .types import ArcaeaOfflineDEFV2_ScoreItem, ScoreExport
def score(score: Score) -> ScoreExport: def score(score: Score) -> ScoreExport:
@ -17,3 +17,21 @@ def score(score: Score) -> ScoreExport:
"clear_type": score.clear_type, "clear_type": score.clear_type,
"comment": score.comment, "comment": score.comment,
} }
def score_def_v2(score: Score) -> ArcaeaOfflineDEFV2_ScoreItem:
return {
"id": score.id,
"songId": score.song_id,
"ratingClass": score.rating_class,
"score": score.score,
"pure": score.pure,
"far": score.far,
"lost": score.lost,
"date": score.date,
"maxRecall": score.max_recall,
"modifier": score.modifier,
"clearType": score.clear_type,
"source": None,
"comment": score.comment,
}

View File

@ -1,4 +1,4 @@
from typing import Optional, TypedDict from typing import List, Literal, Optional, TypedDict
class ScoreExport(TypedDict): class ScoreExport(TypedDict):
@ -14,3 +14,32 @@ class ScoreExport(TypedDict):
modifier: Optional[int] modifier: Optional[int]
clear_type: Optional[int] clear_type: Optional[int]
comment: Optional[str] comment: Optional[str]
class ArcaeaOfflineDEFV2_ScoreItem(TypedDict, total=False):
id: Optional[int]
songId: str
ratingClass: int
score: int
pure: Optional[int]
far: Optional[int]
lost: Optional[int]
date: Optional[int]
maxRecall: Optional[int]
modifier: Optional[int]
clearType: Optional[int]
source: Optional[str]
comment: Optional[str]
ArcaeaOfflineDEFV2_Score = TypedDict(
"ArcaeaOfflineDEFV2_Score",
{
"$schema": Literal[
"https://arcaeaoffline.sevive.xyz/schemas/def/v2/score.schema.json"
],
"type": Literal["score"],
"version": Literal[2],
"scores": List[ArcaeaOfflineDEFV2_ScoreItem],
},
)

View File

@ -0,0 +1 @@
from .b30_csv import SmartRteB30CsvConverter

View File

@ -0,0 +1,64 @@
from sqlalchemy.orm import Session
from ...models import Chart, ScoreBest
from ...utils.rating import rating_class_to_text
class SmartRteB30CsvConverter:
CSV_ROWS = [
"songname",
"songId",
"Difficulty",
"score",
"Perfect",
"criticalPerfect",
"Far",
"Lost",
"Constant",
"singlePTT",
]
def __init__(
self,
session: Session,
):
self.session = session
def rows(self) -> list:
csv_rows = [self.CSV_ROWS.copy()]
with self.session as session:
results = (
session.query(
Chart.title,
ScoreBest.song_id,
ScoreBest.rating_class,
ScoreBest.score,
ScoreBest.pure,
ScoreBest.shiny_pure,
ScoreBest.far,
ScoreBest.lost,
Chart.constant,
ScoreBest.potential,
)
.join(
Chart,
(Chart.song_id == ScoreBest.song_id)
& (Chart.rating_class == ScoreBest.rating_class),
)
.all()
)
for result in results:
# replace the comma in song title because the target project
# cannot handle quoted string
result = list(result)
result[0] = result[0].replace(",", "")
result[2] = rating_class_to_text(result[2])
# divide constant to float
result[-2] = result[-2] / 10
# round potential
result[-1] = round(result[-1], 5)
csv_rows.append(result)
return csv_rows

View File

@ -1,8 +1,12 @@
# pylint: disable=too-few-public-methods
from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm.exc import DetachedInstanceError from sqlalchemy.orm.exc import DetachedInstanceError
class ReprHelper: class ReprHelper:
# pylint: disable=no-member
def _repr(self, **kwargs) -> str: def _repr(self, **kwargs) -> str:
""" """
Helper for __repr__ Helper for __repr__

View File

@ -1,3 +1,5 @@
# pylint: disable=too-few-public-methods
from sqlalchemy import TEXT from sqlalchemy import TEXT
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

View File

@ -1,6 +1,8 @@
# pylint: disable=too-few-public-methods, duplicate-code
from typing import Optional from typing import Optional
from sqlalchemy import TEXT, case, func, inspect, select from sqlalchemy import TEXT, case, func, inspect, select, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy_utils import create_view from sqlalchemy_utils import create_view
@ -37,7 +39,8 @@ class Score(ScoresBase):
comment="0: NORMAL, 1: EASY, 2: HARD" comment="0: NORMAL, 1: EASY, 2: HARD"
) )
clear_type: Mapped[Optional[int]] = mapped_column( clear_type: Mapped[Optional[int]] = mapped_column(
comment="0: TRACK LOST, 1: NORMAL CLEAR, 2: FULL RECALL, 3: PURE MEMORY, 4: EASY CLEAR, 5: HARD CLEAR" comment="0: TRACK LOST, 1: NORMAL CLEAR, 2: FULL RECALL, "
"3: PURE MEMORY, 4: EASY CLEAR, 5: HARD CLEAR"
) )
comment: Mapped[Optional[str]] = mapped_column(TEXT()) comment: Mapped[Optional[str]] = mapped_column(TEXT())
@ -78,10 +81,21 @@ class ScoreCalculated(ScoresViewBase):
Score.score, Score.score,
Score.pure, Score.pure,
( (
Score.score case(
- func.floor( (
(Score.pure * 10000000.0 / ChartInfo.notes) (
+ (Score.far * 0.5 * 10000000.0 / ChartInfo.notes) ChartInfo.notes.is_not(None)
& Score.pure.is_not(None)
& Score.far.is_not(None)
& (ChartInfo.notes != 0)
),
Score.score
- func.floor(
(Score.pure * 10000000.0 / ChartInfo.notes)
+ (Score.far * 0.5 * 10000000.0 / ChartInfo.notes)
),
),
else_=text("NULL"),
) )
).label("shiny_pure"), ).label("shiny_pure"),
Score.far, Score.far,

View File

@ -1,3 +1,5 @@
# pylint: disable=too-few-public-methods, duplicate-code
from typing import Optional from typing import Optional
from sqlalchemy import TEXT, ForeignKey, func, select from sqlalchemy import TEXT, ForeignKey, func, select
@ -154,7 +156,7 @@ class ChartInfo(SongsBase):
ForeignKey("difficulties.rating_class"), primary_key=True ForeignKey("difficulties.rating_class"), primary_key=True
) )
constant: Mapped[int] = mapped_column( constant: Mapped[int] = mapped_column(
comment="real_constant * 10. For example, Crimson Throne [FTR] is 10.4, then store 104 here." comment="real_constant * 10. For example, Crimson Throne [FTR] is 10.4, then store 104."
) )
notes: Mapped[Optional[int]] notes: Mapped[Optional[int]]

View File

@ -103,7 +103,7 @@ class Searcher:
return list(results) return list(results)
def search(self, string: str, *, limit: int = 10, fuzzy_distance: int = 10): def search(self, string: str, *, limit: int = 10):
query_string = f"{string}" query_string = f"{string}"
query = self.default_query_parser.parse(query_string) query = self.default_query_parser.parse(query_string)
with self.index.searcher() as searcher: with self.index.searcher() as searcher:

View File

@ -0,0 +1,15 @@
from datetime import datetime
from enum import IntEnum
class KanaeDayNight(IntEnum):
Day = 0
Night = 1
def kanae_day_night(timestamp: int) -> KanaeDayNight:
"""
:param timestamp: POSIX timestamp, which is passed to `datetime.fromtimestamp(timestamp)`.
"""
dt = datetime.fromtimestamp(timestamp)
return KanaeDayNight.Day if 6 <= dt.hour <= 19 else KanaeDayNight.Night

View File

@ -5,6 +5,7 @@ RATING_CLASS_TEXT_MAP = {
1: "Present", 1: "Present",
2: "Future", 2: "Future",
3: "Beyond", 3: "Beyond",
4: "Eternal",
} }
RATING_CLASS_SHORT_TEXT_MAP = { RATING_CLASS_SHORT_TEXT_MAP = {
@ -12,6 +13,7 @@ RATING_CLASS_SHORT_TEXT_MAP = {
1: "PRS", 1: "PRS",
2: "FTR", 2: "FTR",
3: "BYD", 3: "BYD",
4: "ETR",
} }

View File

@ -2,6 +2,15 @@ from typing import Any, Sequence
SCORE_GRADE_FLOOR = [9900000, 9800000, 9500000, 9200000, 8900000, 8600000, 0] SCORE_GRADE_FLOOR = [9900000, 9800000, 9500000, 9200000, 8900000, 8600000, 0]
SCORE_GRADE_TEXTS = ["EX+", "EX", "AA", "A", "B", "C", "D"] SCORE_GRADE_TEXTS = ["EX+", "EX", "AA", "A", "B", "C", "D"]
MODIFIER_TEXTS = ["NORMAL", "EASY", "HARD"]
CLEAR_TYPE_TEXTS = [
"TRACK LOST",
"NORMAL CLEAR",
"FULL RECALL",
"PURE MEMORY",
"EASY CLEAR",
"HARD CLEAR",
]
def zip_score_grade(score: int, __seq: Sequence, default: Any = "__PRESERVE__"): def zip_score_grade(score: int, __seq: Sequence, default: Any = "__PRESERVE__"):
@ -27,3 +36,11 @@ def zip_score_grade(score: int, __seq: Sequence, default: Any = "__PRESERVE__"):
def score_to_grade_text(score: int) -> str: def score_to_grade_text(score: int) -> str:
return zip_score_grade(score, SCORE_GRADE_TEXTS) return zip_score_grade(score, SCORE_GRADE_TEXTS)
def modifier_to_text(modifier: int) -> str:
return MODIFIER_TEXTS[modifier]
def clear_type_to_text(clear_type: int) -> str:
return CLEAR_TYPE_TEXTS[clear_type]

View File

@ -0,0 +1,22 @@
from decimal import Decimal
from arcaea_offline.calculate.world_step import (
AwakenedAyuPartnerBonus,
LegacyMapStepBooster,
PlayResult,
calculate_step_original,
)
def test_world_step():
# the result was copied from https://arcaea.fandom.com/wiki/World_Mode_Mechanics#Calculation
# CC BY-SA 3.0
booster = LegacyMapStepBooster(6, 250)
partner_bonus = AwakenedAyuPartnerBonus("+3.6")
play_result = PlayResult(play_rating=Decimal("11.299"), partner_step=92)
result = calculate_step_original(
play_result, partner_bonus=partner_bonus, step_booster=booster
)
assert result.quantize(Decimal("0.000")) == Decimal("175.149")

0
tests/db/__init__.py Normal file
View File

5
tests/db/db.py Normal file
View File

@ -0,0 +1,5 @@
from sqlalchemy import Engine, create_engine, inspect
def create_engine_in_memory():
return create_engine("sqlite:///:memory:")

View File

View File

@ -0,0 +1,118 @@
from sqlalchemy import Engine
from sqlalchemy.orm import Session
from arcaea_offline.models.songs import (
Chart,
ChartInfo,
Difficulty,
Pack,
Song,
SongsBase,
SongsViewBase,
)
from ..db import create_engine_in_memory
def _song(**kw):
defaults = {"artist": "test"}
defaults.update(kw)
return Song(**defaults)
def _difficulty(**kw):
defaults = {"rating_plus": False, "audio_override": False, "jacket_override": False}
defaults.update(kw)
return Difficulty(**defaults)
class Test_Chart:
def init_db(self, engine: Engine):
SongsBase.metadata.create_all(engine)
SongsViewBase.metadata.create_all(engine)
def db(self):
db = create_engine_in_memory()
self.init_db(db)
return db
def test_chart_info(self):
pre_entites = [
Pack(id="test", name="Test Pack"),
_song(idx=0, id="song0", set="test", title="Full Chart Info"),
_song(idx=1, id="song1", set="test", title="Partial Chart Info"),
_song(idx=2, id="song2", set="test", title="No Chart Info"),
_difficulty(song_id="song0", rating_class=2, rating=9),
_difficulty(song_id="song1", rating_class=2, rating=9),
_difficulty(song_id="song2", rating_class=2, rating=9),
ChartInfo(song_id="song0", rating_class=2, constant=90, notes=1234),
ChartInfo(song_id="song1", rating_class=2, constant=90),
]
db = self.db()
with Session(db) as session:
session.add_all(pre_entites)
session.commit()
chart_song0_ratingclass2 = (
session.query(Chart)
.where((Chart.song_id == "song0") & (Chart.rating_class == 2))
.one()
)
assert chart_song0_ratingclass2.constant == 90
assert chart_song0_ratingclass2.notes == 1234
chart_song1_ratingclass2 = (
session.query(Chart)
.where((Chart.song_id == "song1") & (Chart.rating_class == 2))
.one()
)
assert chart_song1_ratingclass2.constant == 90
assert chart_song1_ratingclass2.notes is None
chart_song2_ratingclass2 = (
session.query(Chart)
.where((Chart.song_id == "song2") & (Chart.rating_class == 2))
.first()
)
assert chart_song2_ratingclass2 is None
def test_difficulty_title_override(self):
pre_entites = [
Pack(id="test", name="Test Pack"),
_song(idx=0, id="test", set="test", title="Test"),
_difficulty(song_id="test", rating_class=0, rating=2),
_difficulty(song_id="test", rating_class=1, rating=5),
_difficulty(song_id="test", rating_class=2, rating=8),
_difficulty(
song_id="test", rating_class=3, rating=10, title="TEST ~REVIVE~"
),
ChartInfo(song_id="test", rating_class=0, constant=10),
ChartInfo(song_id="test", rating_class=1, constant=10),
ChartInfo(song_id="test", rating_class=2, constant=10),
ChartInfo(song_id="test", rating_class=3, constant=10),
]
db = self.db()
with Session(db) as session:
session.add_all(pre_entites)
session.commit()
charts_original_title = (
session.query(Chart)
.where((Chart.song_id == "test") & (Chart.rating_class in [0, 1, 2]))
.all()
)
assert all(chart.title == "Test" for chart in charts_original_title)
chart_overrided_title = (
session.query(Chart)
.where((Chart.song_id == "test") & (Chart.rating_class == 3))
.one()
)
assert chart_overrided_title.title == "TEST ~REVIVE~"

16
tox.ini Normal file
View File

@ -0,0 +1,16 @@
[tox]
env_list =
py311
py310
py39
py38
minversion = 4.11.3
[testenv]
description = run the tests with pytest
package = wheel
wheel_build_env = .pkg
deps =
pytest==7.4.3
commands =
pytest {tty:--color=yes} {posargs}